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" + ); + } +}