From 9060a37526aec55d4acd81dd67209d601c6f9a6c Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 27 Jul 2026 08:39:29 +0000 Subject: [PATCH] perf(master): skip re-observing experiments whose base version has not moved A scan round fully observed every experiment, every 300s. `observe` lists every MemWAL shard and reads its manifest, opens each flushed generation to count pending rows, and runs `count_rows` over the base table. At the tens of thousands of experiments this deployment is heading for -- the overwhelming majority of them cold and never written again -- a round cannot finish inside its interval: roughly 19 minutes at an optimistic 0.3s per experiment and concurrency 8, and closer to 90 on object storage. The next round starts before the previous one ends, forever. `RolloutObservation` already carried the base dataset version, but `StatRow` never persisted it, so the cheapest possible change signal was being discarded. Opening the dataset reads its manifest, which is where the version comes from, so comparing it against the last observed version costs nothing. An unchanged version means no write has landed, so every derived metric is necessarily unchanged and the previous row can be reused verbatim -- skipping all of the above. Cold experiments now cost one open per round rather than a full observation. Also removes a per-experiment `stats.get()` that took the stats lock once per experiment per round; the previous snapshot is read once up front and passed down. `observe_one` no longer needs `MasterState` at all. Two edges, both tested: `version` is absent from a `_stats` table written before this commit. Reads tolerate that and record `UNKNOWN_VERSION`, which forces one full observation and then persists the column. Erroring instead would break every read on upgrade -- and reads are what the UI and both auto-sweeps depend on, so the master would come up blind until the table happened to be rewritten. `refresh_one` passes `None` so it always observes fully. Its callers use it precisely when they believe something just changed (after a compaction, or an explicit refresh), and the version check would short-circuit exactly the observation they asked for. New gauges `master_scan_experiments_skipped` / `_observed` show how much of each round the version check avoided. A ratio near 1 is the healthy state at scale; a ratio trending to 0 means the incremental path has stopped engaging. Verified the tests are not vacuous: disabling the skip fails two of the four. Co-Authored-By: Claude --- crates/lance-context-master/src/scanner.rs | 307 +++++++++++++++--- crates/lance-context-master/src/scheduler.rs | 2 + .../lance-context-master/src/stats_store.rs | 104 ++++++ 3 files changed, 374 insertions(+), 39 deletions(-) diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index 9f08a6e..6bd0543 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/crates/lance-context-master/src/scanner.rs @@ -173,16 +173,32 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { let live: HashSet = entries.iter().map(|e| e.name.clone()).collect(); let concurrency = state.config.scan_concurrency.max(1); - // Observe experiments concurrently (bounded), preserving prior compaction - // counters by reading the existing stats row first. `None` means this round - // could not observe that experiment; its previous row is carried over below + // Read the previous snapshot once, up front. Each experiment's row supplies + // both its last observed version (to decide whether a full observation is + // needed) and its compaction counters (which are carried forward). This + // used to be a `stats.get()` per experiment, taking the stats lock once per + // experiment per round. + let previous: HashMap = { + let mut stats = state.stats.lock().await; + stats + .list(None, usize::MAX, 0) + .await? + .into_iter() + .map(|row| (row.name.clone(), row)) + .collect() + }; + let previous = Arc::new(previous); + + // Observe experiments concurrently (bounded). `None` means this round could + // not observe that experiment; its previous row is carried over below // rather than dropped. - let observed: Vec<(String, Option)> = stream::iter(entries) + let observed: Vec<(String, Option<(StatRow, bool)>)> = stream::iter(entries) .map(|entry| { - let state = state.clone(); + let previous = previous.clone(); async move { - match observe_one(&state, &entry.name, &entry.uri).await { - Ok(row) => (entry.name, Some(row)), + let prev = previous.get(&entry.name); + match observe_one(&entry.name, &entry.uri, prev).await { + Ok(result) => (entry.name, Some(result)), Err(e) => { tracing::warn!(store = %entry.name, error = %e, "scan: observe failed"); (entry.name, None) @@ -195,6 +211,10 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { .await; let count = observed.iter().filter(|(_, row)| row.is_some()).count(); + let skipped = observed + .iter() + .filter(|(_, row)| matches!(row, Some((_, true)))) + .count(); let mut stats = state.stats.lock().await; @@ -206,16 +226,12 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { // An experiment this round failed to observe keeps its previous row, so a // transient storage error does not blank it from the UI or, worse, hide it // from the compaction and WAL-merge sweeps that read this table. - let previous = stats.list(None, usize::MAX, 0).await?; - let mut carry_over: HashMap = previous - .into_iter() - .map(|row| (row.name.clone(), row)) - .collect(); + let mut carry_over = (*previous).clone(); let mut snapshot: Vec = Vec::with_capacity(observed.len()); for (name, row) in observed { match row { - Some(row) => snapshot.push(row), + Some((row, _skipped)) => snapshot.push(row), None => { if let Some(stale) = carry_over.remove(&name) { snapshot.push(stale); @@ -241,6 +257,18 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { } metrics::histogram!("master_scan_duration_seconds").record(scan_start.elapsed().as_secs_f64()); + // How much of the round was avoided by the version check. A ratio near 1 is + // the healthy state at scale: most experiments are cold and cost only an + // open. A ratio trending to 0 means the incremental path is not engaging -- + // usually a stats table that keeps losing its rows. + metrics::gauge!("master_scan_experiments_skipped").set(skipped as f64); + metrics::gauge!("master_scan_experiments_observed").set((count - skipped) as f64); + tracing::debug!( + total = count, + skipped, + observed = count - skipped, + "scan round complete" + ); // Named without a `_total` suffix: these are gauges (current value), and // `_total` is the counter convention. Datadog's OpenMetrics check infers // type from the name, so `_total` on a gauge was being ingested as a @@ -258,10 +286,29 @@ async fn scan_once_inner(state: &Arc) -> lance::Result { Ok(count) } -/// Open one experiment read-only, observe it, and build its stats row. Preserves -/// `last_compaction`/`total_compactions` from any existing stats row. Returns -/// an error on open/observe failure or timeout. -async fn observe_one(state: &Arc, name: &str, uri: &str) -> lance::Result { +/// Open one experiment read-only and refresh its stats row, skipping the +/// expensive observation when nothing can have changed. +/// +/// `previous` is the row from the last scan, if any. +/// +/// # Why the version check pays for itself +/// +/// Opening the dataset reads its manifest, which already carries the current +/// version — so comparing it against the last observed version is free. If they +/// match, no write has landed since, and every derived metric (`row_count`, +/// `fragment_count`, `pending_wal_generations`) is necessarily unchanged. The +/// previous row is then reused verbatim. +/// +/// What that skips is the whole cost of `observe`: listing every MemWAL shard +/// and reading its manifest, opening each flushed generation to count pending +/// rows, and a `count_rows` over the base table. At tens of thousands of +/// experiments, almost all of them cold, that is the difference between a scan +/// that finishes inside its interval and one that never does. +async fn observe_one( + name: &str, + uri: &str, + previous: Option<&StatRow>, +) -> lance::Result<(StatRow, bool)> { let opts = RolloutStoreOptions::default(); let open = RolloutStore::open_existing_with_options(uri, opts); let store = match tokio::time::timeout(OBSERVE_TIMEOUT, open).await { @@ -273,6 +320,18 @@ async fn observe_one(state: &Arc, name: &str, uri: &str) -> lance:: ))) } }; + + // Free: the manifest is already in hand from the open above. + let current_version = store.version() as i64; + if let Some(prev) = previous { + if prev.version != StatRow::UNKNOWN_VERSION && prev.version == current_version { + let mut row = prev.clone(); + row.uri = uri.to_string(); + row.scanned_at = Utc::now().timestamp_millis(); + return Ok((row, true)); + } + } + let obs = match tokio::time::timeout(OBSERVE_TIMEOUT, store.observe()).await { Ok(Ok(obs)) => obs, Ok(Err(e)) => return Err(e), @@ -283,33 +342,41 @@ async fn observe_one(state: &Arc, name: &str, uri: &str) -> lance:: } }; - // Carry compaction counters forward across scans. - let (last_compaction, total_compactions) = { - let mut stats = state.stats.lock().await; - match stats.get(name).await { - Ok(Some(prev)) => (prev.last_compaction, prev.total_compactions), - _ => (StatRow::NO_COMPACTION, 0), - } - }; - - Ok(StatRow { - name: name.to_string(), - uri: uri.to_string(), - row_count: obs.row_count, - fragment_count: obs.fragment_count, - last_updated: obs.last_updated, - pending_wal_generations: obs.pending_wal_generations, - last_compaction, - total_compactions, - scanned_at: Utc::now().timestamp_millis(), - }) + // Carry compaction counters forward across scans. Taken from the caller's + // snapshot rather than a per-experiment `stats.get()`, which used to take + // the stats lock once per experiment per round. + let (last_compaction, total_compactions) = previous + .map_or((StatRow::NO_COMPACTION, 0), |prev| { + (prev.last_compaction, prev.total_compactions) + }); + + Ok(( + StatRow { + name: name.to_string(), + uri: uri.to_string(), + row_count: obs.row_count, + fragment_count: obs.fragment_count, + last_updated: obs.last_updated, + pending_wal_generations: obs.pending_wal_generations, + last_compaction, + total_compactions, + scanned_at: Utc::now().timestamp_millis(), + version: obs.version as i64, + }, + false, + )) } /// Refresh one experiment immediately and persist its new stats row. +/// +/// Passes `None` as the previous row so this always does a full observation: +/// callers use it precisely when they believe something just changed (after a +/// compaction, or on an explicit refresh request), and the version check would +/// otherwise short-circuit exactly the observation they asked for. pub async fn refresh_one(state: &Arc, name: &str, uri: &str) -> lance::Result<()> { let guard = state.task_store.coordination_lock("stats-writer").await?; - let result = match observe_one(state, name, uri).await { - Ok(row) => state.stats.lock().await.upsert(&row).await, + let result = match observe_one(name, uri, None).await { + Ok((row, _)) => state.stats.lock().await.upsert(&row).await, Err(error) => Err(error), }; let release = state.task_store.release_coordination_lock(guard).await; @@ -415,3 +482,165 @@ mod maintenance_alerting_tests { ); } } + +#[cfg(test)] +mod incremental_scan_tests { + use super::*; + use lance_context_core::{RolloutRecord, RolloutStore, RolloutStoreOptions, ROLE_ASSISTANT}; + use tempfile::TempDir; + + fn rec(id: &str) -> RolloutRecord { + RolloutRecord { + id: id.to_string(), + rollout_id: "r".to_string(), + problem_id: "p".to_string(), + dataset: None, + sequence_order: 0, + role: ROLE_ASSISTANT.to_string(), + created_at: Utc::now(), + content: Some("x".to_string()), + content_type: "text/plain".to_string(), + model_input_string: None, + model_output_string: None, + rationale: None, + problem_text: None, + user_metadata: None, + input_tokens: None, + output_tokens: None, + num_input_tokens: None, + num_output_tokens: None, + output_logprobs: None, + input_logprobs: None, + ref_logprobs: None, + loss_mask: None, + advantage: None, + reward: None, + raw_reward: None, + grader_id: None, + score: None, + include_in_training: None, + exclude_reason: None, + policy_version: None, + relationships: vec![], + binary_payload: None, + payload_size: None, + payload_checksum: None, + artifact_type: None, + metadata: None, + } + } + + /// An experiment whose base version has not moved must not be re-observed. + /// + /// This is the property the whole incremental path rests on: at tens of + /// thousands of experiments, almost all cold, a scan that fully observes + /// every one cannot finish inside its interval. Opening the dataset is + /// unavoidable (it is how the version is read), but everything after it -- + /// listing MemWAL shards, opening each flushed generation, counting rows -- + /// must be skipped when nothing has changed. + #[tokio::test] + async fn unchanged_experiment_is_not_reobserved() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().join("e.lance").to_string_lossy().to_string(); + { + let store = RolloutStore::open_with_options(&uri, RolloutStoreOptions::default()) + .await + .unwrap(); + store.add(&[rec("a")]).await.unwrap(); + store.flush().await.unwrap(); + } + + // First pass: nothing known, so a full observation happens. + let (first, skipped) = observe_one("e", &uri, None).await.unwrap(); + assert!(!skipped, "the first observation cannot be skipped"); + assert_ne!(first.version, StatRow::UNKNOWN_VERSION); + + // Second pass with the row from the first: the version is unchanged, so + // the expensive observation is skipped and the row is reused. + let (second, skipped) = observe_one("e", &uri, Some(&first)).await.unwrap(); + assert!(skipped, "an unchanged experiment must skip re-observation"); + assert_eq!(second.row_count, first.row_count); + assert_eq!(second.fragment_count, first.fragment_count); + assert_eq!(second.version, first.version); + assert!( + second.scanned_at >= first.scanned_at, + "a skipped row still refreshes scanned_at so staleness stays visible" + ); + } + + /// A write moves the base version, so the next scan must observe fully and + /// pick up the new counts. + #[tokio::test] + async fn changed_experiment_is_reobserved() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().join("e.lance").to_string_lossy().to_string(); + let mut store = RolloutStore::open_with_options(&uri, RolloutStoreOptions::default()) + .await + .unwrap(); + store.add(&[rec("a")]).await.unwrap(); + store.flush().await.unwrap(); + + let (first, _) = observe_one("e", &uri, None).await.unwrap(); + + // Merge the WAL into the base table: this advances the base version. + store.cleanup_own_shard().await.unwrap(); + + let (second, skipped) = observe_one("e", &uri, Some(&first)).await.unwrap(); + assert!( + !skipped, + "a changed base version must force a full observation" + ); + assert_ne!( + second.version, first.version, + "the new row must record the new version" + ); + } + + /// A row from before the `version` column existed carries the sentinel and + /// must force one full observation rather than being treated as unchanged. + #[tokio::test] + async fn unknown_version_forces_observation() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().join("e.lance").to_string_lossy().to_string(); + { + let store = RolloutStore::open_with_options(&uri, RolloutStoreOptions::default()) + .await + .unwrap(); + store.add(&[rec("a")]).await.unwrap(); + store.flush().await.unwrap(); + } + + let (mut legacy, _) = observe_one("e", &uri, None).await.unwrap(); + legacy.version = StatRow::UNKNOWN_VERSION; + + let (refreshed, skipped) = observe_one("e", &uri, Some(&legacy)).await.unwrap(); + assert!( + !skipped, + "an unknown version must not be mistaken for an unchanged one" + ); + assert_ne!(refreshed.version, StatRow::UNKNOWN_VERSION); + } + + /// Compaction counters survive a skipped round. + #[tokio::test] + async fn skipped_round_preserves_compaction_counters() { + let dir = TempDir::new().unwrap(); + let uri = dir.path().join("e.lance").to_string_lossy().to_string(); + { + let store = RolloutStore::open_with_options(&uri, RolloutStoreOptions::default()) + .await + .unwrap(); + store.add(&[rec("a")]).await.unwrap(); + store.flush().await.unwrap(); + } + + let (mut prev, _) = observe_one("e", &uri, None).await.unwrap(); + prev.last_compaction = 1_700_000_000_000; + prev.total_compactions = 7; + + let (next, skipped) = observe_one("e", &uri, Some(&prev)).await.unwrap(); + assert!(skipped); + assert_eq!(next.last_compaction, 1_700_000_000_000); + assert_eq!(next.total_compactions, 7); + } +} diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index b1fecfb..19c7b0b 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -347,6 +347,7 @@ async fn update_stats_after_compaction(state: &Arc, name: &str, sto last_compaction: Utc::now().timestamp_millis(), total_compactions: prev_total + 1, scanned_at: Utc::now().timestamp_millis(), + version: obs.version as i64, }; if let Err(e) = stats.upsert(&row).await { tracing::warn!(store = %name, error = %e, "post-compaction stats upsert failed"); @@ -833,6 +834,7 @@ mod tests { let state = MasterState::new(cfg).await.unwrap(); let seed = |name: &str, pending: i64| StatRow { + version: StatRow::UNKNOWN_VERSION, name: name.to_string(), uri: state.rollout_uri(name), row_count: 0, diff --git a/crates/lance-context-master/src/stats_store.rs b/crates/lance-context-master/src/stats_store.rs index c8688ad..5ce9ba7 100644 --- a/crates/lance-context-master/src/stats_store.rs +++ b/crates/lance-context-master/src/stats_store.rs @@ -40,12 +40,26 @@ pub struct StatRow { pub last_compaction: i64, pub total_compactions: i64, pub scanned_at: i64, + /// Base-dataset version at the last full observation. + /// + /// A scan compares this against the experiment's current version to decide + /// whether anything can have changed. Opening the dataset already reads the + /// manifest, so the current version is free; an unchanged version means the + /// expensive parts of `observe` (listing MemWAL shards, opening every + /// flushed generation, counting rows) can be skipped entirely. `-1` marks a + /// row written before this column existed, forcing one full observation. + pub version: i64, } impl StatRow { /// Sentinel stored in `last_compaction` when no compaction has run. pub const NO_COMPACTION: i64 = -1; + /// Sentinel stored in `version` for a row written before that column + /// existed, or whose version could not be read. Never matches a real + /// version, so it always forces a full observation. + pub const UNKNOWN_VERSION: i64 = -1; + /// Convert to the wire DTO, mapping the `-1` sentinel back to `None`. #[must_use] pub fn into_summary(self) -> ExperimentSummary { @@ -84,6 +98,7 @@ fn stats_schema() -> Schema { Field::new("last_compaction", DataType::Int64, false), Field::new("total_compactions", DataType::Int64, false), Field::new("scanned_at", DataType::Int64, false), + Field::new("version", DataType::Int64, false), ]) } @@ -168,6 +183,7 @@ impl StatsStore { Arc::new(Int64Array::from(vec![row.last_compaction])), Arc::new(Int64Array::from(vec![row.total_compactions])), Arc::new(Int64Array::from(vec![row.scanned_at])), + Arc::new(Int64Array::from(vec![row.version])), ], )?) } @@ -206,6 +222,9 @@ impl StatsStore { Arc::new(Int64Array::from( rows.iter().map(|r| r.scanned_at).collect::>(), )), + Arc::new(Int64Array::from( + rows.iter().map(|r| r.version).collect::>(), + )), ], )?) } @@ -412,6 +431,12 @@ impl StatsStore { let last_compaction = Self::i64_col(batch, "last_compaction")?; let total_compactions = Self::i64_col(batch, "total_compactions")?; let scanned_at = Self::i64_col(batch, "scanned_at")?; + // Absent on a table written before this column existed. Reading must + // tolerate that rather than erroring, or an upgrade breaks every + // deployment until the table happens to be rewritten. Missing means + // "version unknown", which forces one full observation and then + // persists the column. + let version = Self::i64_col(batch, "version").ok(); let mut out = Vec::with_capacity(batch.num_rows()); for i in 0..batch.num_rows() { out.push(StatRow { @@ -424,6 +449,7 @@ impl StatsStore { last_compaction: last_compaction.value(i), total_compactions: total_compactions.value(i), scanned_at: scanned_at.value(i), + version: version.map_or(StatRow::UNKNOWN_VERSION, |v| v.value(i)), }); } Ok(out) @@ -525,6 +551,7 @@ mod tests { pending_wal_generations: 0, last_compaction: StatRow::NO_COMPACTION, total_compactions: 0, + version: 1, scanned_at: 1_700_000_001_000, } } @@ -660,6 +687,7 @@ mod scale_tests { pending_wal_generations: 0, last_compaction: StatRow::NO_COMPACTION, total_compactions: 0, + version: 1, scanned_at: 1_700_000_000_000, } } @@ -823,3 +851,79 @@ mod scale_tests { ); } } + +#[cfg(test)] +mod migration_tests { + use super::*; + use arrow_schema::{DataType, Field, Schema as ArrowSchema}; + use tempfile::TempDir; + + /// A `_stats` table written before the `version` column existed must still + /// be readable, and the first snapshot must migrate it forward. + /// + /// Without this, an upgrade fails on every read until the table happens to + /// be rewritten -- and the read path is what the UI and both auto-sweeps + /// use, so the master would come up blind. + #[tokio::test] + async fn table_without_version_column_is_readable_and_migrates() { + let dir = TempDir::new().unwrap(); + let uri = dir + .path() + .join("_stats.lance") + .to_string_lossy() + .to_string(); + + // Build a table with the pre-`version` schema. + let legacy_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("name", DataType::Utf8, false), + Field::new("uri", DataType::Utf8, false), + Field::new("row_count", DataType::Int64, false), + Field::new("fragment_count", DataType::Int64, false), + Field::new("last_updated", DataType::Int64, false), + Field::new("pending_wal_generations", DataType::Int64, false), + Field::new("last_compaction", DataType::Int64, false), + Field::new("total_compactions", DataType::Int64, false), + Field::new("scanned_at", DataType::Int64, false), + ])); + let batch = RecordBatch::try_new( + legacy_schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["old-exp"])), + Arc::new(StringArray::from(vec!["/data/old-exp.lance"])), + Arc::new(Int64Array::from(vec![42i64])), + Arc::new(Int64Array::from(vec![7i64])), + Arc::new(Int64Array::from(vec![1_700_000_000_000i64])), + Arc::new(Int64Array::from(vec![3i64])), + Arc::new(Int64Array::from(vec![-1i64])), + Arc::new(Int64Array::from(vec![0i64])), + Arc::new(Int64Array::from(vec![1_700_000_000_000i64])), + ], + ) + .unwrap(); + let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), legacy_schema); + Dataset::write(reader, &uri, None).await.unwrap(); + + // Reading must succeed and report the sentinel, not error. + let mut s = StatsStore::open_or_create(&uri, None).await.unwrap(); + let rows = s.list(None, 10, 0).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].name, "old-exp"); + assert_eq!( + rows[0].version, + StatRow::UNKNOWN_VERSION, + "a pre-migration row must read as version-unknown, forcing one full observation" + ); + assert_eq!(rows[0].row_count, 42, "existing data must survive"); + + // The first snapshot rewrites the table with the new schema. + let mut migrated = rows[0].clone(); + migrated.version = 11; + s.write_snapshot(&[migrated]).await.unwrap(); + + let after = s.list(None, 10, 0).await.unwrap(); + assert_eq!( + after[0].version, 11, + "the snapshot must persist the version" + ); + } +}