diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs index dba46fd9..b026aeda 100644 --- a/crates/lance-context-master/src/catchup/executor.rs +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -1,13 +1,14 @@ use super::Result; use crate::{config::MasterConfig, maintenance_execution, state::MasterState}; +use futures::future::BoxFuture; use lance_context_api::TaskKind; use lance_context_core::{ GenericStore, GenericStoreOptions, MergeMemoryBudget, RolloutStore, RolloutStoreOptions, }; use lance_context_merge::MaintenanceKind; -use std::{sync::Arc, time::Duration}; +use std::{future::Future, sync::Arc, time::Duration}; -/// One dedicated process, one table, serial shards, bounded passes. No server, +/// One dedicated process, one table, ordered commits, bounded passes. No server, /// scanner, ingestion writer, or fleet task consumer starts in this mode. pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { config.catchup.enabled = false; @@ -90,75 +91,364 @@ async fn merge_passes(state: &Arc, target: &str) -> Result // request another Job. Visit every shard before repeating any hot shard. for _ in 0..16 { let before = total; - for shard in &config.shards { - if started.elapsed() >= admit_for { - return Ok(format!( - "dedicated catch-up slice merged {total} generations" - )); + total += run_pipeline(&config.shards, config.pipeline_enabled, |shard| { + let session = session.clone(); + let budget = budget.clone(); + async move { + // Admission is checked after queue capacity becomes available. + // A prepared batch already admitted may finish beyond the slice. + if started.elapsed() >= admit_for { + return Ok(Box::pin(async { Ok(0usize) }) as BoxFuture<'_, Result>); + } + let read_started = tokio::time::Instant::now(); + let commit = prepare_shard(state, target, shard, session, budget).await?; + let prepare_secs = read_started.elapsed().as_secs_f64(); + Ok(Box::pin(async move { + let commit_started = tokio::time::Instant::now(); + let reclaimed = commit.await?; + let commit_secs = commit_started.elapsed().as_secs_f64(); + metrics::counter!("master_catchup_generations_reclaimed_total") + .increment(reclaimed as u64); + tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs, + "dedicated catch-up committed"); + Ok(reclaimed) + }) as BoxFuture<'_, Result>) } - let reclaimed = if let Some(name) = target.strip_prefix("generic:") { - let options = GenericStoreOptions { - shard_id: Some(shard.clone()), - merge_after_generations: Some(0), - merge_max_generations: Some(8), - merge_max_bytes: Some(config.merge_max_bytes), - key_index_type: state.config.key_index_type, - merge_budget: Some(budget.clone()), - session: Some(session.clone()), - pending_generations_max: Some(0), - ..Default::default() - }; - let mut store = GenericStore::open_existing(&state.generic_uri(name), options) + }) + .await?; + if total == before { + break; + } + } + Ok(format!("dedicated catch-up merged {total} generations")) +} + +/// Read sealed WAL without acquiring a writer epoch. The returned future owns +/// the prepared data and its memory reservation; it performs no writes until +/// the single consumer polls it under the existing maintenance write scope. +async fn prepare_shard( + state: &Arc, + target: &str, + shard: &str, + session: Arc, + budget: Arc, +) -> Result>> { + if let Some(name) = target.strip_prefix("generic:") { + let options = GenericStoreOptions { + shard_id: Some(shard.to_owned()), + merge_after_generations: Some(0), + merge_max_generations: Some(state.config.catchup.merge_max_generations), + merge_max_bytes: Some(state.config.catchup.merge_max_bytes), + key_index_type: state.config.key_index_type, + merge_budget: Some(budget.clone()), + session: Some(session.clone()), + pending_generations_max: Some(0), + ..Default::default() + }; + let mut store = GenericStore::open_existing(&state.generic_uri(name), options) + .await + .map_err(|e| e.to_string())?; + // No resident writer was opened, so cleanup only sees sealed WAL. + match store + .prepare_cleanup_merge() + .await + .map_err(|e| e.to_string())? + { + Some((manifest_store, manifest, prepared)) => Ok(Box::pin(async move { + store + .commit_prepared_merge(&manifest_store, &manifest, prepared) .await - .map_err(|e| e.to_string())?; - // No resident writer was opened, so cleanup only sees sealed WAL. - match store - .prepare_cleanup_merge() + .map_err(|e| e.to_string()) + })), + None => Ok(Box::pin(async { Ok(0usize) })), + } + } else { + let options = RolloutStoreOptions { + shard_id: Some(shard.to_owned()), + merge_after_generations: Some(0), + merge_max_generations: Some(state.config.catchup.merge_max_generations), + merge_max_bytes: Some(state.config.catchup.merge_max_bytes), + key_index_type: state.config.key_index_type, + merge_budget: Some(budget.clone()), + session: Some(session.clone()), + pending_generations_max: Some(0), + ..Default::default() + }; + let mut store = + RolloutStore::open_existing_with_options(&state.rollout_uri(target), options) + .await + .map_err(|e| e.to_string())?; + match store + .prepare_merge_if_ready(1) + .await + .map_err(|e| e.to_string())? + { + Some((manifest_store, manifest, prepared)) => Ok(Box::pin(async move { + store + .commit_prepared_merge(&manifest_store, &manifest, prepared) .await - .map_err(|e| e.to_string())? - { - Some((manifest_store, manifest, prepared)) => store - .commit_prepared_merge(&manifest_store, &manifest, prepared) - .await - .map_err(|e| e.to_string())?, - None => 0, - } - } else { - let options = RolloutStoreOptions { - shard_id: Some(shard.clone()), - merge_after_generations: Some(0), - merge_max_generations: Some(8), - merge_max_bytes: Some(config.merge_max_bytes), - key_index_type: state.config.key_index_type, - merge_budget: Some(budget.clone()), - session: Some(session.clone()), - pending_generations_max: Some(0), - ..Default::default() - }; - let mut store = - RolloutStore::open_existing_with_options(&state.rollout_uri(target), options) - .await - .map_err(|e| e.to_string())?; - match store - .prepare_merge_if_ready(1) + .map_err(|e| e.to_string()) + })), + None => Ok(Box::pin(async { Ok(0usize) })), + } + } +} + +// StorageBase releases the failed generation/reservation before returning this +// error. A speculative reader can encounter it while the committer holds bytes. +const MERGE_BUDGET_BUSY: &str = "merge memory budget busy while reading first generation; retry"; + +/// At most one commit plus one preparing/queued batch. Reserve queue capacity +/// BEFORE preparing: otherwise a producer can retain a third batch. Do not use +/// `buffered(n)` here: a later read holding the entire memory budget can block +/// an earlier read forever while ordered delivery prevents its commit. +/// +/// Both halves are polled in the caller's task, preserving the task-local +/// maintenance fence and progress scope. Cancellation drops both halves and +/// their reservations; already admitted manifest writes are drained by the scope. +async fn run_pipeline(items: I, enabled: bool, mut prepare: P) -> Result +where + I: IntoIterator, + I::Item: Clone, + P: FnMut(I::Item) -> F, + F: Future>, + C: Future>, +{ + if !enabled { + let mut total = 0; + for item in items { + total += prepare(item).await?.await?; + } + return Ok(total); + } + let (send, mut receive) = tokio::sync::mpsc::channel::>(1); + let (completed, mut completion) = tokio::sync::watch::channel(0usize); + let producer = async move { + for (sent, item) in items.into_iter().enumerate() { + let permit = send + .reserve() + .await + .map_err(|_| "catch-up committer stopped")?; + let mut prepared = prepare(item.clone()).await; + if prepared + .as_ref() + .is_err_and(|error| error.contains(MERGE_BUDGET_BUSY)) + { + // Finish the preceding commit before retrying this shard alone. + // Do not prepare later shards while it waits: they could take + // the freed memory and recreate the contention. Retry only once; + // storage/other failures retain the normal durable backoff. + let finished = completion + .wait_for(|count| *count >= sent) .await - .map_err(|e| e.to_string())? - { - Some((manifest_store, manifest, prepared)) => store - .commit_prepared_merge(&manifest_store, &manifest, prepared) - .await - .map_err(|e| e.to_string())?, - None => 0, - } - }; - total += reclaimed; - metrics::counter!("master_catchup_generations_reclaimed_total") - .increment(reclaimed as u64); - tracing::info!(%target, %shard, reclaimed, total, "dedicated catch-up committed"); + .map_err(|_| "catch-up committer stopped before memory retry")?; + drop(finished); + metrics::counter!("master_catchup_prefetch_memory_retries_total").increment(1); + prepared = prepare(item).await; + } + let failed = prepared.is_err(); + permit.send(prepared); + // Deliver read errors in order: let the preceding commit finish, + // then stop. Never keep preparing later shards after a read failure. + if failed { + break; + } } - if total == before { - break; + Ok::<_, String>(()) + }; + let consumer = async move { + let mut total = 0; + let mut finished = 0; + while let Some(prepared) = receive.recv().await { + total += prepared?.await?; + finished += 1; + completed.send_replace(finished); } + Ok::<_, String>(total) + }; + let (_, total) = tokio::try_join!(producer, consumer)?; + Ok(total) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::{Notify, Semaphore}; + + #[tokio::test] + async fn pipeline_overlaps_reads_with_one_ordered_commit_and_bounds_lookahead() { + let prepared = AtomicUsize::new(0); + let committed = std::sync::Mutex::new(Vec::new()); + let next_ready = Notify::new(); + let release = Semaphore::new(0); + let work = run_pipeline(0..5, true, |id| { + let prepared = &prepared; + let committed = &committed; + let next_ready = &next_ready; + let release = &release; + async move { + prepared.fetch_add(1, Ordering::SeqCst); + if id == 1 { + next_ready.notify_one(); + } + Ok(async move { + if id == 0 { + release.acquire().await.unwrap().forget(); + } + committed.lock().unwrap().push(id); + Ok(1) + }) + } + }); + let check = async { + next_ready.notified().await; + assert_eq!(prepared.load(Ordering::SeqCst), 2); + assert!(committed.lock().unwrap().is_empty()); + release.add_permits(1); + }; + let (result, ()) = + tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(work, check) }) + .await + .unwrap(); + assert_eq!(result.unwrap(), 5); + assert_eq!(*committed.lock().unwrap(), [0, 1, 2, 3, 4]); + } + + #[tokio::test] + async fn pipeline_progresses_when_only_one_batch_fits_and_releases_on_failure() { + for enabled in [false, true] { + for fail_at in [None, Some(2)] { + let budget = MergeMemoryBudget::new(1024 * 1024); + let committed = AtomicUsize::new(0); + let result = tokio::time::timeout( + Duration::from_secs(5), + run_pipeline(0..5, enabled, |id| { + let budget = &budget; + let committed = &committed; + async move { + let reservation = budget.reserve(budget.limit()).await; + Ok(async move { + let _reservation = reservation; + tokio::task::yield_now().await; + if fail_at == Some(id) { + return Err("commit failed".into()); + } + committed.fetch_add(1, Ordering::SeqCst); + Ok(1) + }) + } + }), + ) + .await + .expect("a queued reservation must not block the committer"); + assert_eq!( + result, + fail_at.map_or(Ok(5), |_| Err("commit failed".into())) + ); + assert_eq!(committed.load(Ordering::SeqCst), fail_at.unwrap_or(5)); + assert_eq!(budget.reserved(), 0); + } + } + } + + #[tokio::test] + async fn oversized_prefetch_retries_after_commit_releases_memory() { + let budget = MergeMemoryBudget::new(2 * 1024 * 1024); + let attempts = AtomicUsize::new(0); + let contention = Notify::new(); + let result = tokio::time::timeout( + Duration::from_secs(5), + run_pipeline(0..2, true, |id| { + let budget = &budget; + let attempts = &attempts; + let contention = &contention; + async move { + attempts.fetch_add(1, Ordering::SeqCst); + let mut reservation = budget.reserve(1024 * 1024).await; + if id == 1 && !reservation.try_grow_to(2 * 1024 * 1024) { + contention.notify_one(); + return Err(MERGE_BUDGET_BUSY.into()); + } + Ok(async move { + let _reservation = reservation; + if id == 0 { + // Force the first attempt at shard 1 to contend with + // this commit, then permit the commit to release bytes. + contention.notified().await; + } + Ok(1) + }) + } + }), + ) + .await + .unwrap(); + assert_eq!(result, Ok(2)); + assert_eq!(attempts.load(Ordering::SeqCst), 3); + assert_eq!(budget.reserved(), 0); + } + + #[tokio::test] + async fn read_failure_finishes_previous_commit_and_stops_admission() { + let prepared = AtomicUsize::new(0); + let committed = AtomicUsize::new(0); + let failed_read = Notify::new(); + let result = tokio::time::timeout( + Duration::from_secs(5), + run_pipeline(0..5, true, |id| { + let prepared = &prepared; + let committed = &committed; + let failed_read = &failed_read; + async move { + prepared.fetch_add(1, Ordering::SeqCst); + if id == 1 { + failed_read.notify_one(); + return Err("read failed".into()); + } + Ok(async move { + failed_read.notified().await; + committed.fetch_add(1, Ordering::SeqCst); + Ok(1) + }) + } + }), + ) + .await + .unwrap(); + assert_eq!(result, Err("read failed".into())); + assert_eq!(prepared.load(Ordering::SeqCst), 2); + assert_eq!(committed.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn cancellation_drops_active_and_prefetched_reservations() { + let budget = MergeMemoryBudget::new(2 * 1024 * 1024); + let next_ready = Notify::new(); + let mut work = Box::pin(run_pipeline(0..5, true, |id| { + let budget = &budget; + let next_ready = &next_ready; + async move { + let reservation = budget.reserve(1024 * 1024).await; + if id == 1 { + next_ready.notify_one(); + } + Ok(async move { + let _reservation = reservation; + std::future::pending::>().await + }) + } + })); + tokio::time::timeout(Duration::from_secs(5), async { + tokio::select! { + result = &mut work => panic!("unexpected completion: {result:?}"), + () = next_ready.notified() => {} + } + }) + .await + .unwrap(); + assert_eq!(budget.reserved(), budget.limit()); + drop(work); + assert_eq!(budget.reserved(), 0); } - Ok(format!("dedicated catch-up merged {total} generations")) } diff --git a/crates/lance-context-master/src/catchup/kubernetes.rs b/crates/lance-context-master/src/catchup/kubernetes.rs index fbb10425..b068144c 100644 --- a/crates/lance-context-master/src/catchup/kubernetes.rs +++ b/crates/lance-context-master/src/catchup/kubernetes.rs @@ -242,9 +242,17 @@ pub(super) fn render_job(config: &MasterConfig, record: &Record, mut spec: Value ("ETCD_ENDPOINTS", config.etcd.etcd_endpoints.join(",")), ("ETCD_PREFIX", config.etcd.etcd_prefix.clone()), ("CATCHUP_ENABLED", "false".into()), + ( + "CATCHUP_PIPELINE_ENABLED", + config.catchup.pipeline_enabled.to_string(), + ), ("CATCHUP_TARGET", record.target.clone()), ("CATCHUP_JOB_NAME", record.job.clone()), ("CATCHUP_SHARDS", shards.join(",")), + ( + "CATCHUP_MERGE_MAX_GENERATIONS", + config.catchup.merge_max_generations.to_string(), + ), ( "CATCHUP_MERGE_MAX_BYTES", config.catchup.merge_max_bytes.to_string(), diff --git a/crates/lance-context-master/src/catchup/mod.rs b/crates/lance-context-master/src/catchup/mod.rs index c6d91df3..5bdeb880 100644 --- a/crates/lance-context-master/src/catchup/mod.rs +++ b/crates/lance-context-master/src/catchup/mod.rs @@ -45,6 +45,9 @@ pub struct CatchupConfig { /// Stable writer identities, not Pod IPs. Only sealed generations are merged. #[arg(long, env = "CATCHUP_SHARDS", value_delimiter = ',')] pub shards: Vec, + /// Generation count cap in addition to the byte cap; zero disables this cap. + #[arg(long, env = "CATCHUP_MERGE_MAX_GENERATIONS", default_value_t = 64)] + pub merge_max_generations: usize, #[arg(long, env = "CATCHUP_MERGE_MAX_BYTES", default_value_t = 67_108_864)] pub merge_max_bytes: usize, #[arg( @@ -53,6 +56,9 @@ pub struct CatchupConfig { default_value_t = 1_073_741_824 )] pub merge_memory_bytes: usize, + /// Prepare the next shard while the current shard commits. Commits stay ordered. + #[arg(long, env = "CATCHUP_PIPELINE_ENABLED", default_value_t = true, action = clap::ArgAction::Set)] + pub pipeline_enabled: bool, /// Native one-table executor mode; does not start the admin server or scanner. #[arg(long, env = "CATCHUP_TARGET", hide = true)] pub target: Option, @@ -72,8 +78,10 @@ impl Default for CatchupConfig { slice_secs: 1800, startup_timeout_secs: 1800, shards: Vec::new(), + merge_max_generations: 64, merge_max_bytes: 67_108_864, merge_memory_bytes: 1_073_741_824, + pipeline_enabled: true, target: None, job_name: None, } diff --git a/crates/lance-context-master/src/catchup/tests.rs b/crates/lance-context-master/src/catchup/tests.rs index 8f357765..14106521 100644 --- a/crates/lance-context-master/src/catchup/tests.rs +++ b/crates/lance-context-master/src/catchup/tests.rs @@ -54,6 +54,8 @@ fn admission_requires_fresh_owned_pressure_even_for_manual_requests() { fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { let mut c = config(); c.key_index_type = lance_context_core::KeyIndexType::Zonemap; + c.catchup.pipeline_enabled = false; + c.catchup.merge_max_generations = 32; let r = Record { target: "hot".into(), job: "lc-catchup-test".into(), @@ -75,6 +77,8 @@ fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { let mut pod = template(); pod["containers"][0]["env"] = json!([ {"name":"CATCHUP_ENABLED","value":"true"}, + {"name":"CATCHUP_PIPELINE_ENABLED","value":"true"}, + {"name":"CATCHUP_MERGE_MAX_GENERATIONS","value":"8"}, {"name":"ROLLOUT_KEY_INDEX_TYPE","value":"btree"} ]); pod["activeDeadlineSeconds"] = json!(1); @@ -93,6 +97,14 @@ fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { let env = job["spec"]["template"]["spec"]["containers"][0]["env"] .as_array() .unwrap(); + for (name, expected) in [ + ("CATCHUP_PIPELINE_ENABLED", "false"), + ("CATCHUP_MERGE_MAX_GENERATIONS", "32"), + ] { + let matches: Vec<_> = env.iter().filter(|e| e["name"] == name).collect(); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0]["value"], expected); + } assert_eq!( env.iter() .filter(|e| e["name"] == "ROLLOUT_KEY_INDEX_TYPE") @@ -273,6 +285,12 @@ async fn policy_disagreement_cannot_expand_the_cluster_budget() { #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn native_executor_preserves_live_ingestion_and_drains_sealed_shards() { + for memory_bytes in [1024 * 1024, 2 * 1024 * 1024, 3 * 1024 * 1024] { + native_executor_with_budget(memory_bytes).await; + } +} + +async fn native_executor_with_budget(memory_bytes: usize) { use lance_context_core::{ ColumnSpec, ColumnType, GenericStore, GenericStoreOptions, SchemaSpec, }; @@ -303,11 +321,21 @@ async fn native_executor_preserves_live_ingestion_and_drains_sealed_shards() { .await .unwrap(); for generation in 0..4 { - let row = json!({"id":format!("{shard}-{generation}"),"text":"x".repeat(512 * 1024)}) + let payload_bytes = if memory_bytes == 3 * 1024 * 1024 && shard == "worker-1" { + 2 * 1024 * 1024 + } else { + 512 * 1024 + }; + let row = + json!({"id":format!("{shard}-{generation}"),"text":"x".repeat(payload_bytes)}) + .as_object() + .unwrap() + .clone(); + let shared = json!({"id":"shared", "text":format!("{shard}-{generation}")}) .as_object() .unwrap() .clone(); - writer.add(&[row]).await.unwrap(); + writer.add(&[row, shared]).await.unwrap(); writer.flush().await.unwrap(); } writers.push(writer); @@ -315,6 +343,10 @@ async fn native_executor_preserves_live_ingestion_and_drains_sealed_shards() { assert_eq!(writers[0].pending_wal_generations().await.unwrap(), 8); let mut cfg = state.config.clone(); cfg.catchup.enabled = false; + // Exercise both overlapping reads and a budget that admits one batch. + // Prefetch must not block the commit that releases its reservation. + cfg.catchup.merge_max_bytes = 512 * 1024; + cfg.catchup.merge_memory_bytes = memory_bytes; cfg.merge_rollout.owned_targets.push(target.into()); let decision = Inventory::new(&state) .reserve(target, "integration", 8, 100) @@ -357,10 +389,14 @@ async fn native_executor_preserves_live_ingestion_and_drains_sealed_shards() { .unwrap(); assert!(reader.pending_wal_generations().await.unwrap() <= 18); let rows = reader.list(None, None).await.unwrap(); - assert_eq!(rows.len(), 26); + assert_eq!(rows.len(), 27); + assert_eq!( + rows.iter().find(|r| r["id"] == "shared").unwrap()["text"], + "worker-1-3" + ); let ids: std::collections::HashSet<_> = rows.iter().map(|r| r["id"].as_str().unwrap()).collect(); - assert_eq!(ids.len(), 26); + assert_eq!(ids.len(), 27); assert!(state .task_store .merge_coordinator() @@ -502,3 +538,108 @@ async fn stale_progress_revoke_cannot_cancel_advancing_or_replaced_execution() { .await .unwrap(); } + +/// Full native executor comparison; setup and verification are outside timing. +/// Opt in explicitly so CI's ignored etcd suite does not run a benchmark. +#[tokio::test] +#[ignore = "manual benchmark: CATCHUP_BENCH=1 ETCD_TEST_ENDPOINTS required"] +async fn benchmark_native_catchup_pipeline() { + if std::env::var("CATCHUP_BENCH").as_deref() != Ok("1") { + return; + } + use lance_context_core::{ + ColumnSpec, ColumnType, GenericStore, GenericStoreOptions, SchemaSpec, + }; + let workload = std::env::var("CATCHUP_BENCH_WORKLOAD").unwrap_or_else(|_| "updates".into()); + assert!(["updates", "unique"].contains(&workload.as_str())); + for (pipeline, max_generations) in [ + (false, 8), + (true, 8), + (true, 64), + (true, 64), + (true, 8), + (false, 8), + ] { + let (_dir, state) = fixture().await.unwrap(); + let target = "generic:hot"; + let uri = state.generic_uri("hot"); + let spec = SchemaSpec::new(vec![ + ( + "id".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ( + "text".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ]); + let shards: Vec<_> = (0..4).map(|i| format!("worker-{i}")).collect(); + let mut writers = Vec::new(); + for shard in &shards { + let writer = GenericStore::open( + &uri, + spec.clone(), + GenericStoreOptions { + shard_id: Some(shard.clone()), + merge_after_generations: Some(0), + session: Some(lance_context_core::RolloutStore::build_session( + 32 << 20, + 32 << 20, + )), + ..Default::default() + }, + ) + .await + .unwrap(); + for generation in 0..32 { + let rows: Vec<_> = (0..8).map(|id| { + json!({"id": if workload == "updates" { format!("{shard}-{id}") } else { format!("{shard}-{generation}-{id}") }, "text": format!("{generation}:{}", "x".repeat(8192))}) + .as_object().unwrap().clone() + }).collect(); + writer.add(&rows).await.unwrap(); + writer.flush().await.unwrap(); + } + writers.push(writer); + } + let pending = writers[0].pending_wal_generations().await.unwrap(); + assert_eq!(pending, 128); + let mut cfg = state.config.clone(); + cfg.catchup.enabled = false; + cfg.catchup.shards = shards; + cfg.catchup.pipeline_enabled = pipeline; + cfg.catchup.merge_max_generations = max_generations; + cfg.merge_rollout.owned_targets.push(target.into()); + cfg.catchup.target = Some(target.into()); + cfg.catchup.job_name = Inventory::new(&state) + .reserve(target, "benchmark", pending as i64, 100) + .await + .unwrap() + .job; + let started = std::time::Instant::now(); + execute(cfg, target).await.unwrap(); + let seconds = started.elapsed().as_secs_f64(); + let reader = GenericStore::open_existing(&uri, GenericStoreOptions::default()) + .await + .unwrap(); + assert_eq!(reader.pending_wal_generations().await.unwrap(), 0); + let rows = reader.list(None, None).await.unwrap(); + let expected_rows = if workload == "updates" { 32 } else { 1024 }; + assert_eq!(rows.len(), expected_rows); + for row in rows { + let generation = if workload == "updates" { + "31" + } else { + row["id"].as_str().unwrap().split('-').nth(2).unwrap() + }; + assert_eq!(row["text"], format!("{generation}:{}", "x".repeat(8192))); + } + let base = lance::Dataset::open(&uri).await.unwrap(); + assert_eq!(base.count_rows(None).await.unwrap(), expected_rows); + eprintln!( + "CATCHUP_BENCH {}", + json!({"workload":workload,"pipeline":pipeline,"max_generations":max_generations, + "seconds":seconds,"generations":pending,"generations_per_second":pending as f64 / seconds, + "base_version":base.version().version}) + ); + } +} diff --git a/docs/master-catchup.md b/docs/master-catchup.md index c7427882..68da0ee9 100644 --- a/docs/master-catchup.md +++ b/docs/master-catchup.md @@ -40,14 +40,29 @@ still apply. It does not enable owned targets or restart production workers. Defaults: four Jobs cluster-wide, 256 pending generations, a 30-second scan interval, and stats no older than 900 seconds. A Job has one container, one table, -serial shard visits, no Kubernetes retries, a 1800-second soft admission slice, a +ordered shard commits, no Kubernetes retries, a 1800-second soft admission slice, a 30-second termination grace period, and a 24-hour terminal retention period. The Job cannot preempt other Pods. CPU/memory requests must equal explicit limits. The sample reserves 2 CPUs and 8 GiB per Job (8 CPUs/32 GiB total at the default fleet cap); size these against available capacity before enabling. -The native process uses a 128 MiB shared Lance cache, one merge at a time, -eight generations and 64 MiB per batch, and a 1 GiB merge-buffer budget. These +The native process uses a 128 MiB shared Lance cache, a single committer, +64 generations (`CATCHUP_MERGE_MAX_GENERATIONS`) and 64 MiB per batch, and a +1 GiB merge-buffer budget. Generation count and byte limits both apply; zero +count disables only the count cap. Small generations can now share a commit +instead of repeating index maintenance every eight generations. + +`CATCHUP_PIPELINE_ENABLED=true` (default) overlaps the next shard's preparation +with the current shard's commit. At most two batches are preparing/retained, +sharing the same memory budget. Commit and drain order remains shard order; +no ordinary worker fan-out is made concurrent. A budget that only admits one +batch falls back naturally to serial reads/commits. If an indivisible generation +cannot grow its reservation while the preceding commit holds memory, prefetch +waits for that commit and retries the same shard once with the freed budget. +Later shards cannot consume those bytes first. Other read failures retain normal +failure/backoff behavior. Set the flag to `false` and +the generation cap to `8` for a comparison with the previous executor. Both +settings are passed explicitly to new Jobs. These are buffering bounds, not a total RSS guarantee: one indivisible generation and Lance working memory can exceed the buffer bound. Kubernetes limits provide the final process bound. Validate real large generations before increasing concurrency.