diff --git a/crates/lance-context-core/src/datagen_store.rs b/crates/lance-context-core/src/datagen_store.rs index 37b289bc..aa46a468 100644 --- a/crates/lance-context-core/src/datagen_store.rs +++ b/crates/lance-context-core/src/datagen_store.rs @@ -59,6 +59,11 @@ pub struct DatagenStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Periodically merge this writer's pending generations. `None` or zero /// disables the timer. pub cleanup_interval_secs: Option, @@ -103,6 +108,7 @@ impl DatagenStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: None, schema: Arc::new(datagen_log_schema()), // Datagen keys on `event_id`, not `id`: event ids are derived diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 873df0ec..5a202a6b 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -71,6 +71,12 @@ pub struct GenericStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. Every pending generation is a + /// separate dataset a read must open, so this is the read-amplification + /// alarm. `None` uses the crate default (256); `Some(0)` disables the warn. + /// The `rollout_wal_pending_generations` histogram is emitted regardless. + pub pending_generations_warn: Option, /// Shared, capacity-bounded Lance session. pub session: Option>, /// Whether [`GenericStore::add`] seals before returning, making the rows it @@ -174,6 +180,7 @@ impl GenericStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: options.session, schema: create_schema, // Always `id`: the LSM merge key, which `SchemaSpec::validate` @@ -794,6 +801,54 @@ mod tests { }); } + /// Every LSM read samples each shard's pending-generation count. This is + /// the signal that would have caught a store at 12k pending generations + /// days before it OOMKilled its readers, so a read must emit it. + #[test] + fn reads_sample_pending_generations_per_shard() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::MetricKind; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let rt = tokio::runtime::Runtime::new().unwrap(); + + metrics::with_local_recorder(&recorder, || { + rt.block_on(async { + let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap(); + // Three sealed adds leave three flushed generations pending on + // this instance's single shard. + for i in 0..3 { + store + .add(&[row(json!({"id": format!("r{i}")}))]) + .await + .unwrap(); + } + store.list(None, None).await.unwrap(); + }); + }); + + let samples: Vec = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| { + key.kind() == MetricKind::Histogram + && key.key().name() == crate::metrics::ROLLOUT_WAL_PENDING_GENERATIONS + }) + .flat_map(|(_, _, _, value)| match value { + DebugValue::Histogram(v) => v.into_iter().map(|s| s.into_inner()).collect(), + _ => Vec::new(), + }) + .collect(); + assert!( + samples.contains(&3.0), + "a read must sample the shard's 3 pending generations; got {samples:?}" + ); + } + #[test] fn duplicate_ids_dedup_to_the_newest_write() { // The LSM merge key is `id`, so a re-added id supersedes the old row. diff --git a/crates/lance-context-core/src/metrics.rs b/crates/lance-context-core/src/metrics.rs index 332f6ca6..2ee86eef 100644 --- a/crates/lance-context-core/src/metrics.rs +++ b/crates/lance-context-core/src/metrics.rs @@ -54,6 +54,17 @@ pub const ROLLOUT_WAL_MERGE_DURATION: &str = "rollout_wal_merge_duration_seconds /// aborts on the first failing phase, so this also identifies where it died. pub const ROLLOUT_WAL_MERGE_ERRORS: &str = "rollout_wal_merge_errors_total"; +/// Flushed MemWAL generations pending merge on one shard, sampled once per +/// shard on every LSM read (`wal_shard_snapshots`). Unlabelled: the store and +/// shard go on the accompanying `warn!` span, not on the series. +/// +/// This is the read-amplification signal. Every pending generation is a +/// separate dataset the read path must open, so a shard whose count climbs into +/// the hundreds means its owner's merge has stopped keeping up — the failure +/// that took a worker fleet down at ~12k pending generations across 20 shards +/// before it was noticed in crash logs. +pub const ROLLOUT_WAL_PENDING_GENERATIONS: &str = "rollout_wal_pending_generations"; + /// Emit a histogram sample in seconds. No-op without the `metrics` feature. #[cfg(feature = "metrics")] macro_rules! observe_duration { @@ -69,6 +80,22 @@ macro_rules! observe_duration { }}; } +/// Emit a histogram sample of a plain value (a count, not a duration). No-op +/// without the `metrics` feature. +#[cfg(feature = "metrics")] +macro_rules! observe_value { + ($name:expr, $value:expr $(, $k:expr => $v:expr)* $(,)?) => { + ::metrics::histogram!($name $(, $k => $v)*).record($value as f64) + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! observe_value { + ($name:expr, $value:expr $(, $k:expr => $v:expr)* $(,)?) => {{ + let _ = &$value; + }}; +} + /// Increment a counter by one. No-op without the `metrics` feature. /// /// The disabled arm still expands to a unit *expression* rather than an empty @@ -144,4 +171,6 @@ macro_rules! observe_phase { }}; } -pub(crate) use {count, observe_duration, observe_phase, timer_elapsed, timer_start}; +pub(crate) use { + count, observe_duration, observe_phase, observe_value, timer_elapsed, timer_start, +}; diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index cf5e6202..2082cef4 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -383,6 +383,11 @@ pub struct RolloutStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Shared Lance [`Session`] used to open this store's base dataset (and, /// transitively, every flushed MemWAL generation it reads — those inherit /// the base dataset's session). @@ -469,6 +474,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, } = options; let mut base = StorageBase::open( @@ -479,6 +485,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema: Arc::new(rollout_schema()), key_column: "id".to_string(), @@ -2487,6 +2494,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, schema: legacy_schema.clone(), key_column: "id".to_string(), @@ -2806,6 +2814,7 @@ mod tests { merge_after_generations: Some(0), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -2850,6 +2859,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }; { @@ -2899,6 +2909,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -2950,6 +2961,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }; let instance_a = RolloutStore::open_with_options(&uri, options("rollout-0")) @@ -3171,6 +3183,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -3233,6 +3246,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -3333,6 +3347,7 @@ mod tests { merge_after_generations: None, // no merge → epoch never reclaimed merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3391,6 +3406,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3430,6 +3446,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3566,6 +3583,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3652,6 +3670,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3699,6 +3718,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3768,6 +3788,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3805,6 +3826,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3863,6 +3885,7 @@ mod tests { merge_after_generations: Some(2), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3904,6 +3927,7 @@ mod tests { merge_after_generations: None, // count trigger off merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -4403,6 +4427,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -4501,6 +4526,7 @@ mod tests { merge_after_generations: None, // disabled merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -4533,6 +4559,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await diff --git a/crates/lance-context-core/src/store.rs b/crates/lance-context-core/src/store.rs index ca6c313f..3dfc022e 100644 --- a/crates/lance-context-core/src/store.rs +++ b/crates/lance-context-core/src/store.rs @@ -302,6 +302,11 @@ pub struct ContextStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Whether [`ContextStore::add`] seals the memtable before returning, so the /// rows it wrote are immediately readable. /// @@ -333,6 +338,7 @@ impl Default for ContextStoreOptions { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, // Read-your-write by default; see the field docs. seal_on_add: true, } @@ -614,6 +620,7 @@ impl ContextStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: None, schema: Arc::new(arrow_schema.clone()), key_column: "id".to_string(), @@ -2237,6 +2244,7 @@ impl ContextStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, // A compactor never appends, so the seal mode is irrelevant to it; // deferring keeps it from ever emitting a generation. seal_on_add: false, diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 0e75933c..22b893d8 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -67,7 +67,9 @@ use lance_index::IndexType; use tracing::{info, warn}; use uuid::Uuid; -use crate::metrics::{count, observe_duration, observe_phase, timer_elapsed, timer_start}; +use crate::metrics::{ + count, observe_duration, observe_phase, observe_value, timer_elapsed, timer_start, +}; use crate::store::{CompactionConfig, CompactionStats}; /// Number of shard manifest files to scan per batch when discovering the latest @@ -97,6 +99,14 @@ pub(crate) const DEFAULT_MERGE_MAX_GENERATIONS: usize = 8; /// Buffered Arrow array bytes per merge pass, checked at generation boundaries. pub(crate) const DEFAULT_MERGE_MAX_BYTES: usize = 1024 * 1024 * 1024; +/// Default per-shard pending-generation count at which an LSM read warns. +/// +/// Sized well above the count trigger (`merge_after_generations`, 50 in the +/// reference deployment) and the per-pass cap (`merge_max_generations`, 8), so a +/// healthy store never trips it, while sitting far below the counts that make a +/// read unaffordable: at ~1,500 per shard, reads exhausted a 32 GiB worker. +pub(crate) const DEFAULT_PENDING_GENERATIONS_WARN: usize = 256; + /// Execute only the first `max_source_fragments` from a Lance compaction plan. /// /// Lance's built-in `max_source_fragments` stops before a whole planned task @@ -226,6 +236,10 @@ pub(crate) struct StorageBaseOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when a single shard has at least this many flushed generations + /// pending merge (sampled on every LSM read). `None` uses the crate default + /// (256); `Some(0)` disables the warning. The metric is always emitted. + pub pending_generations_warn: Option, /// Shared, capacity-bounded Lance session. `None` preserves Lance's /// per-open default (a fresh 6 GiB index + 1 GiB metadata session *per /// store*, which is the source of unbounded per-append RSS growth). @@ -287,6 +301,8 @@ pub(crate) struct StorageBase { merge_after_generations: usize, merge_max_generations: usize, merge_max_bytes: usize, + /// Per-shard pending-generation count at which reads warn; `0` disables. + pending_generations_warn: usize, /// Timestamp of the last successful [`Self::compact`] on this handle. last_compaction: Option>, /// Number of successful compactions performed by this handle. @@ -343,6 +359,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -373,6 +390,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -397,6 +415,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -422,6 +441,8 @@ impl StorageBase { merge_after_generations: merge_after_generations.unwrap_or(0), merge_max_generations: merge_max_generations.unwrap_or(DEFAULT_MERGE_MAX_GENERATIONS), merge_max_bytes: merge_max_bytes.unwrap_or(DEFAULT_MERGE_MAX_BYTES), + pending_generations_warn: pending_generations_warn + .unwrap_or(DEFAULT_PENDING_GENERATIONS_WARN), last_compaction: None, total_compactions: 0, last_compaction_error: None, @@ -1386,11 +1407,14 @@ impl StorageBase { let object_store = self.dataset.object_store(None).await?; let branch_path = self.dataset.branch_location().path.clone(); let shard_ids = self.dataset.list_mem_wal_latest_shard_ids().await?; + let warn_at = self.pending_generations_warn; + let uri: Arc = Arc::from(self.dataset.uri()); let snapshots: Vec> = stream::iter(shard_ids) .map(|shard_id| { let object_store = object_store.clone(); let branch_path = branch_path.clone(); + let uri = uri.clone(); async move { let manifest_store = ShardManifestStore::new( object_store, @@ -1401,6 +1425,21 @@ impl StorageBase { let Some(manifest) = manifest_store.read_latest().await? else { return Ok(None); }; + let pending = manifest.flushed_generations.len(); + observe_value!(crate::metrics::ROLLOUT_WAL_PENDING_GENERATIONS, pending); + if warn_at != 0 && pending >= warn_at { + // Every pending generation is one more dataset this read + // must open; past this point the shard's owner has stopped + // keeping up and reads are on their way to exhausting memory. + warn!( + uri = %uri, + shard = %shard_id, + pending, + warn_at, + "MemWAL shard has many flushed generations pending merge; \ + reads open every one of them" + ); + } let mut snapshot = ShardSnapshot::new(shard_id) .with_spec_id(manifest.shard_spec_id) .with_current_generation(manifest.current_generation); diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index 3c54d9c2..bf0b2919 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -62,6 +62,15 @@ const JOB_LATENCY_METRICS: &[&str] = &[ "rollout_wal_merge_lock_wait_seconds", ]; +/// Buckets (upper bounds, generation counts) for `rollout_wal_pending_generations`. +/// +/// Powers of two from the per-pass merge cap (8) up past the point where a read +/// is unaffordable (~1,500 per shard exhausted a 32 GiB worker), so both the +/// healthy band and the runaway band resolve. 9 buckets, same cardinality +/// reasoning as [`REQUEST_LATENCY_BUCKETS`]. +const PENDING_GENERATIONS_BUCKETS: &[f64] = + &[8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1024.0, 4096.0]; + /// Handle used to render the Prometheus exposition text on demand, plus a /// process-resource collector that is refreshed on each scrape. #[derive(Clone)] @@ -92,6 +101,12 @@ pub fn install_recorder() -> MetricsHandle { .set_buckets_for_metric(Matcher::Full((*name).to_string()), JOB_LATENCY_BUCKETS) .expect("job latency buckets are non-empty"); } + builder = builder + .set_buckets_for_metric( + Matcher::Full("rollout_wal_pending_generations".to_string()), + PENDING_GENERATIONS_BUCKETS, + ) + .expect("pending generation buckets are non-empty"); let prometheus = builder .install_recorder() .expect("failed to install Prometheus recorder"); diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index b4ad4e4a..c4eb86b5 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -50,6 +50,18 @@ pub struct ServerConfig { #[arg(long, env = "ROLLOUT_MERGE_MAX_BYTES", default_value = "1073741824")] pub rollout_merge_max_bytes: usize, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every LSM read of any store kind. Every + /// pending generation is a separate dataset a read must open, so this is + /// the read-amplification alarm; `0` disables the warn. The + /// `rollout_wal_pending_generations` histogram is emitted regardless. + #[arg( + long, + env = "ROLLOUT_WAL_PENDING_WARN_GENERATIONS", + default_value = "256" + )] + pub rollout_wal_pending_warn_generations: usize, + /// Interval, in seconds, for the periodic per-shard WAL cleanup task. When /// non-zero, the global sweeper folds this instance's flushed MemWAL /// generations into the base table on a schedule — the *time* half of the @@ -213,6 +225,7 @@ mod tests { let config = ServerConfig::try_parse_from(["lance-context-server", "--data-dir", "/tmp/x"]) .expect("the documented minimal invocation must parse"); assert_eq!(config.rollout_merge_max_bytes, 1024 * 1024 * 1024); + assert_eq!(config.rollout_wal_pending_warn_generations, 256); assert_eq!(config.rollout_flush_interval_secs, 30); assert_eq!(config.rollout_cleanup_interval_secs, 0); } diff --git a/crates/lance-context-server/src/routes/datagen.rs b/crates/lance-context-server/src/routes/datagen.rs index bc6f1281..5895d519 100644 --- a/crates/lance-context-server/src/routes/datagen.rs +++ b/crates/lance-context-server/src/routes/datagen.rs @@ -48,6 +48,7 @@ pub async fn create_datagen_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), cleanup_interval_secs: None, }; let store = DatagenStore::open_with_options(&uri, options) diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index 69ccda2d..2e1cf153 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -59,6 +59,7 @@ pub async fn create_generic_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), session: None, seal_on_add: req.seal_on_add, }; diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index 0d27dba0..c9c9a160 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -199,6 +199,7 @@ pub async fn create_rollout_store( .then_some(state.rollout_merge_after_generations), merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), session: state.rollout_session.clone(), }; diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index ef6d15fb..5fd09292 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -121,6 +121,8 @@ pub struct AppState { pub rollout_merge_max_generations: usize, /// Buffered Arrow array byte budget per merge pass; `0` disables this cap. pub rollout_merge_max_bytes: usize, + /// Per-shard pending-generation count at which reads warn; `0` disables. + pub rollout_wal_pending_warn_generations: usize, /// Periodic per-shard WAL-cleanup interval in seconds; `0` disables the /// global sweeper. See [`Self::spawn_global_sweeper`]. pub rollout_cleanup_interval_secs: u64, @@ -312,6 +314,7 @@ impl AppState { rollout_merge_after_generations: config.rollout_merge_after_generations, rollout_merge_max_generations: config.rollout_merge_max_generations, rollout_merge_max_bytes: config.rollout_merge_max_bytes, + rollout_wal_pending_warn_generations: config.rollout_wal_pending_warn_generations, rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs, rollout_flush_interval_secs: config.rollout_flush_interval_secs, blob_budget, @@ -377,6 +380,7 @@ impl AppState { rollout_merge_after_generations: 0, rollout_merge_max_generations: 8, rollout_merge_max_bytes: 1024 * 1024 * 1024, + rollout_wal_pending_warn_generations: 256, rollout_cleanup_interval_secs: 0, rollout_flush_interval_secs: 0, blob_budget: None, @@ -412,6 +416,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), session: self.rollout_session.clone(), } } @@ -600,6 +605,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), cleanup_interval_secs: None, } } @@ -742,6 +748,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), session: self.rollout_session.clone(), seal_on_add, } diff --git a/crates/lance-context/src/unified_datagen.rs b/crates/lance-context/src/unified_datagen.rs index bf1615a2..1668bbce 100644 --- a/crates/lance-context/src/unified_datagen.rs +++ b/crates/lance-context/src/unified_datagen.rs @@ -38,6 +38,7 @@ impl DatagenStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, cleanup_interval_secs: None, }; let store = LocalStore::open_with_options(uri, options) diff --git a/crates/lance-context/src/unified_generic.rs b/crates/lance-context/src/unified_generic.rs index 92685542..70ba02ae 100644 --- a/crates/lance-context/src/unified_generic.rs +++ b/crates/lance-context/src/unified_generic.rs @@ -41,6 +41,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, seal_on_add, }; @@ -62,6 +63,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, seal_on_add, };