diff --git a/crates/lance-context-core/src/rollout_append.rs b/crates/lance-context-core/src/rollout_append.rs index cf8aacea..29ff4708 100644 --- a/crates/lance-context-core/src/rollout_append.rs +++ b/crates/lance-context-core/src/rollout_append.rs @@ -360,6 +360,37 @@ pub(crate) async fn reconcile_shard(dataset: &Dataset, shard: Uuid) -> Result>, + shards: &[String], +) -> Result { + let dataset = StorageBase::load_with_options(uri, None, session).await?; + dataset_has_sealed_wal(&dataset, shards).await +} + +async fn dataset_has_sealed_wal(dataset: &Dataset, shards: &[String]) -> Result { + if shards.is_empty() || shards.len() > 256 { + return Err(Error::invalid_input( + "expected 1..=256 WAL shard identities", + )); + } + let mut seen = HashSet::new(); + for name in shards { + let id = derive_shard_id(Some(name)); + if seen.insert(id) { + if let Some(manifest) = shard_store(dataset, id).await?.refresh_latest().await? { + if !manifest.sstables.is_empty() { + return Ok(true); + } + } + } + } + Ok(false) +} + /// The only publisher. Open this inside the existing MergeWriteScope so both /// table commits and shard drains retain ownership checks and version fencing. pub struct AppendCoordinator { @@ -395,27 +426,7 @@ impl AppendCoordinator { /// No writer opens, raw WAL replay, generation payload reads, or publication. /// Committed-but-undrained generations also need a maintenance pass. pub async fn has_sealed_wal(&self, shards: &[String]) -> Result { - if shards.is_empty() || shards.len() > 256 { - return Err(Error::invalid_input( - "expected 1..=256 WAL shard identities", - )); - } - let mut seen = HashSet::new(); - for name in shards { - let id = derive_shard_id(Some(name)); - if seen.insert(id) { - if let Some(manifest) = shard_store(&self.dataset, id) - .await? - .refresh_latest() - .await? - { - if !manifest.sstables.is_empty() { - return Ok(true); - } - } - } - } - Ok(false) + dataset_has_sealed_wal(&self.dataset, shards).await } pub async fn plan( diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs index aac1e0d1..b22502ca 100644 --- a/crates/lance-context-master/src/catchup/executor.rs +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -120,16 +120,37 @@ async fn merge_passes(state: &Arc, target: &str) -> Result if state.config.append.enabled(target) { return crate::rollout_append::run(state, target).await; } - let config = &state.config.catchup; let session = RolloutStore::build_session(96 * 1024 * 1024, 32 * 1024 * 1024); - let budget = MergeMemoryBudget::new(config.merge_memory_bytes); + let budget = MergeMemoryBudget::new(state.config.catchup.merge_memory_bytes); + merge_local(state, target, session, budget).await +} + +/// Same bounded storage implementation for resident masters and dedicated legacy executors. +/// Resources are supplied by the caller so resident tasks share one process budget. +pub(crate) async fn merge_local( + state: &Arc, + target: &str, + session: Arc, + budget: Arc, +) -> Result { + let config = &state.config.catchup; let mut total = 0usize; + // Legacy supervisors consume these progress messages and task details. + let dedicated = config.target.is_some(); + let detail = |total| { + let prefix = if dedicated { + "dedicated catch-up merged" + } else { + "local WAL merge reclaimed" + }; + format!("{prefix} {total} generations") + }; let started = tokio::time::Instant::now(); // This is only an admission time slice. A progressing operation already // admitted may run beyond it; cancellation is based on lack of progress. let admit_for = Duration::from_secs(config.slice_secs); - // Bound each slice even under continuous ingestion; later fresh stats may - // request another Job. Visit every shard before repeating any hot shard. + // Bound each slice even under continuous ingestion. Resident masters publish + // continuation demand before releasing their claim; no full scan is required. for _ in 0..16 { let before = total; total += run_pipeline(&config.shards, config.pipeline_enabled, |shard| { @@ -150,18 +171,37 @@ async fn merge_passes(state: &Arc, target: &str) -> Result 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"); + if dedicated { + tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs, + "dedicated catch-up committed"); + } else { + tracing::info!(%target, %shard, reclaimed, prepare_secs, commit_secs, + "local WAL merge committed"); + } Ok(reclaimed) }) as BoxFuture<'_, Result>) } }) .await?; - if total == before { + if total == before && (total == 0 || started.elapsed() < admit_for) { + // Empty metadata reads can consume the admission slice too. They + // must not create a self-sustaining chain of zero-work tasks; + // bounded resident discovery will reassert any remaining demand. + return Ok(detail(total)); + } + if started.elapsed() >= admit_for { break; } } - Ok(format!("dedicated catch-up merged {total} generations")) + if crate::local_merge::selected(&state.config, target) && config.target.is_none() { + state + .task_store + .merge_coordinator() + .request_merge(target) + .await?; + metrics::counter!("master_resident_merge_continuations_total").increment(1); + } + Ok(detail(total)) } /// Read sealed WAL without acquiring a writer epoch. The returned future owns diff --git a/crates/lance-context-master/src/catchup/mod.rs b/crates/lance-context-master/src/catchup/mod.rs index edee8f36..2f11d1ab 100644 --- a/crates/lance-context-master/src/catchup/mod.rs +++ b/crates/lance-context-master/src/catchup/mod.rs @@ -12,6 +12,7 @@ use axum::{ extract::{Query, State}, Json, }; +pub(crate) use executor::merge_local; pub use executor::{execute, ExecuteOutcome}; use kubernetes::Kubernetes; use serde::{Deserialize, Serialize}; diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index bf37e9d8..0852416f 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -15,6 +15,8 @@ pub struct MasterConfig { #[command(flatten)] pub append: crate::rollout_append::AppendConfig, #[command(flatten)] + pub local_merge: crate::local_merge::LocalMergeConfig, + #[command(flatten)] pub wal_tail: crate::wal_tail::WalTailConfig, #[command(flatten)] pub planner: crate::planner::PlannerConfig, diff --git a/crates/lance-context-master/src/eligibility.rs b/crates/lance-context-master/src/eligibility.rs index 8a5a293b..e6019d40 100644 --- a/crates/lance-context-master/src/eligibility.rs +++ b/crates/lance-context-master/src/eligibility.rs @@ -17,8 +17,7 @@ use lance_context_merge::rollout::MergeRollout; pub enum Skip { /// The poller's kind set does not include this task kind. KindNotRunnable, - /// A resident-only poller saw a task that is not a local append merge - /// on an owned, non-draining, opted-in rollout target. + /// A resident-only poller saw a task outside the owned, non-draining local scope. NotResidentEligible, /// A prepared compaction or index is waiting for its commit turn on /// this table; merge yields one admission turn. @@ -90,10 +89,10 @@ pub fn pre_dependency_filter( } if let Some(targets) = resident_targets { let eligible = kind == TaskKind::MergeWal - && !is_generic(target) && policy.rollout.owned(target) && !policy.rollout.draining(target) - && listed(targets, target); + && (targets.iter().any(|t| t == target) + || (!is_generic(target) && listed(targets, target))); if !eligible { return Err(Skip::NotResidentEligible); } @@ -232,10 +231,7 @@ mod tests { check(TaskKind::MergeWal, "excluded"), Err(Skip::NotResidentEligible) ); - assert_eq!( - check(TaskKind::MergeWal, "generic:gs"), - Err(Skip::NotResidentEligible) - ); + assert_eq!(check(TaskKind::MergeWal, "generic:gs"), Ok(())); assert_eq!(check(TaskKind::Compact, "hot"), Err(Skip::KindNotRunnable)); let wildcard = strings(&["*"]); assert_eq!( diff --git a/crates/lance-context-master/src/executors.rs b/crates/lance-context-master/src/executors.rs index 4a849c3b..bdf49fb2 100644 --- a/crates/lance-context-master/src/executors.rs +++ b/crates/lance-context-master/src/executors.rs @@ -213,7 +213,7 @@ pub(crate) fn local_heartbeat(state: &MasterState) -> Heartbeat { cfg.merge_wal_concurrency } else { cfg.task_concurrency.max(1) - } + if cfg.append.rollout_append_local { + } + if crate::local_merge::enabled(cfg) { cfg.append.rollout_append_local_task_concurrency } else { 0 @@ -222,7 +222,7 @@ pub(crate) fn local_heartbeat(state: &MasterState) -> Heartbeat { kinds.push(TaskKind::MergeWal); slots_total.insert(TaskKind::MergeWal, merge_slots as u32); } - let bytes_total = if cfg.append.rollout_append_local { + let bytes_total = if crate::local_merge::enabled(cfg) { cfg.append.rollout_append_local_memory_bytes as u64 } else { 0 diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs index 47fa122a..ab85630d 100644 --- a/crates/lance-context-master/src/lib.rs +++ b/crates/lance-context-master/src/lib.rs @@ -11,6 +11,7 @@ pub mod discovery; pub mod eligibility; pub mod error; pub mod executors; +pub mod local_merge; mod maintenance_execution; mod merge_execution; mod resident_recovery; diff --git a/crates/lance-context-master/src/local_merge.rs b/crates/lance-context-master/src/local_merge.rs new file mode 100644 index 00000000..ab1277f9 --- /dev/null +++ b/crates/lance-context-master/src/local_merge.rs @@ -0,0 +1,137 @@ +//! Native maintenance execution inside each master, sharing its local buffer budget. +use crate::{config::MasterConfig, state::MasterState}; +use std::sync::Arc; + +#[derive(Clone, Debug, Default, clap::Args)] +pub struct LocalMergeConfig { + /// Execute sealed WAL merges in the master, including generic upsert tables. + /// '*' selects only explicitly owned targets; it never migrates legacy writers. + #[arg(long, env = "MASTER_LOCAL_MERGE_TARGETS", value_delimiter = ',')] + pub master_local_merge_targets: Vec, +} + +impl LocalMergeConfig { + pub(crate) fn selected(&self, target: &str) -> bool { + self.master_local_merge_targets + .iter() + .any(|t| t == "*" || t == target) + } +} + +pub(crate) fn enabled(config: &MasterConfig) -> bool { + config.append.rollout_append_local || !config.local_merge.master_local_merge_targets.is_empty() +} + +pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool { + config.merge_rollout.owned(target) + && !config.merge_rollout.draining(target) + && (config.append.local(target) || config.local_merge.selected(target)) +} + +/// Resolve the local selection before claiming. In particular an append wildcard +/// cannot accidentally reserve a slot for a generic table that still needs RPCs. +pub(crate) fn targets(config: &MasterConfig) -> Vec { + config + .merge_rollout + .owned_targets + .iter() + .filter(|target| selected(config, target)) + .cloned() + .collect() +} + +pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> { + if config.local_merge.master_local_merge_targets.is_empty() { + return Ok(()); + } + let catchup = &config.catchup; + if catchup.enabled || catchup.pod_template.is_some() || catchup.target.is_some() { + return Err("master local merges cannot be combined with Kubernetes catch-up mode".into()); + } + if catchup.shards.is_empty() + || catchup.shards.len() > 256 + || catchup.shards.iter().any(|s| s.is_empty()) + || catchup.slice_secs == 0 + || catchup.merge_max_bytes == 0 + || config.append.rollout_append_local_memory_bytes < catchup.merge_max_bytes + || config.merge_wal_concurrency == 0 + { + return Err("master local merges require stable shards, bounded passes and a sufficient shared memory budget".into()); + } + if targets(config) + .iter() + .any(|target| config.append.enabled(target)) + { + config.append.validate()?; + if config.append.rollout_append_local_memory_bytes < config.append.rollout_append_max_bytes + { + return Err("local append batch exceeds shared memory budget".into()); + } + } + for target in &config.local_merge.master_local_merge_targets { + if target != "*" && !selected(config, target) { + return Err(format!( + "master local merge requires a non-draining owned target: {target}" + )); + } + } + Ok(()) +} + +/// The caller owns the canonical task and maintenance fence. No private queue, +/// Pod, writer epoch, or independent per-table memory budget is created here. +pub(crate) async fn run(state: &Arc, target: &str) -> Result { + if !selected(&state.config, target) { + return Err("target is not configured for master local merge".into()); + } + let local = state + .local_append + .as_ref() + .ok_or("missing shared local merge resources")?; + crate::catchup::merge_local(state, target, local.session.clone(), local.budget.clone()).await +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + + #[test] + fn local_selection_preserves_legacy_and_append_boundaries() { + let mut config = MasterConfig::parse_from(["test"]); + config.merge_rollout.owned_targets = vec!["rollout".into(), "generic:rows".into()]; + config.append.rollout_append_local = true; + config.append.rollout_append_targets = vec!["*".into()]; + assert_eq!(targets(&config), ["rollout"]); + config.local_merge.master_local_merge_targets = vec!["*".into()]; + assert_eq!(targets(&config), ["rollout", "generic:rows"]); + assert!(!selected(&config, "legacy")); + config.merge_rollout.drain_targets = vec!["generic:rows".into()]; + assert_eq!(targets(&config), ["rollout"]); + } + + #[test] + fn local_validation_requires_safe_resources_and_ownership() { + let mut config = MasterConfig::parse_from(["test"]); + config.local_merge.master_local_merge_targets = vec!["generic:rows".into()]; + config.catchup.shards = vec!["writer".into()]; + assert!(validate(&config).is_err()); + config.merge_rollout.owned_targets = vec!["generic:rows".into()]; + assert!(validate(&config).is_ok()); + config.append.rollout_append_local_memory_bytes = config.catchup.merge_max_bytes - 1; + assert!(validate(&config).is_err()); + config.append.rollout_append_local_memory_bytes = config.catchup.merge_max_bytes; + config.catchup.enabled = true; + assert!(validate(&config).is_err()); + config.catchup.enabled = false; + config.catchup.shards.clear(); + assert!(validate(&config).is_err()); + config.catchup.shards = vec!["writer".into()]; + config.local_merge.master_local_merge_targets = vec!["*".into()]; + config.merge_rollout.owned_targets.push("rollout".into()); + config.append.rollout_append_targets = vec!["rollout".into()]; + config.append.rollout_append_max_bytes = + config.append.rollout_append_local_memory_bytes + 1; + assert!(validate(&config).is_err()); + } +} diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index 5bcd488e..3ff7aa61 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -32,8 +32,8 @@ pub(crate) async fn run_merge_wal( claim: &TaskClaim, ) -> Result { if state.config.worker_endpoints.is_empty() - && !(state.config.append.local(&claim.task.target) - && state.config.merge_rollout.owned(&claim.task.target)) + && !state.config.merge_rollout.draining(&claim.task.target) + && !crate::local_merge::selected(&state.config, &claim.task.target) { return Err("no worker endpoints configured".into()); } @@ -62,7 +62,7 @@ pub(crate) async fn run_merge_wal( } return run_legacy(state, target).await; } - if state.config.append.enabled(target) { + if state.config.append.enabled(target) || crate::local_merge::selected(&state.config, target) { if let Some(old) = coordinator.get(target).await? { ensure_recovery_due(&coordinator, &old).await?; recover_execution(state, &coordinator, &proof, old).await?; @@ -71,7 +71,13 @@ pub(crate) async fn run_merge_wal( state, claim, lance_context_merge::MaintenanceKind::Catchup, - crate::rollout_append::run(state, target), + async { + if state.config.append.enabled(target) { + crate::rollout_append::run(state, target).await + } else { + crate::local_merge::run(state, target).await + } + }, ) .await; } diff --git a/crates/lance-context-master/src/resident_recovery.rs b/crates/lance-context-master/src/resident_recovery.rs index 82cac85d..2916ac87 100644 --- a/crates/lance-context-master/src/resident_recovery.rs +++ b/crates/lance-context-master/src/resident_recovery.rs @@ -4,7 +4,7 @@ use crate::{config::MasterConfig, state::MasterState, stats_store::StatRow}; use etcd_client::{Compare, CompareOp, Txn, TxnOp}; use futures::{stream, StreamExt}; use lance_context_api::TaskKind; -use lance_context_core::rollout_append::AppendCoordinator; +use lance_context_core::rollout_append::has_sealed_wal; use serde::{Deserialize, Serialize}; use std::{ hash::{Hash, Hasher}, @@ -29,17 +29,19 @@ fn scope(config: &MasterConfig) -> u64 { &config.merge_rollout.owned_targets, &config.merge_rollout.drain_targets, &config.append.rollout_append_targets, + &config.local_merge.master_local_merge_targets, ] { let mut sorted = names.clone(); sorted.sort_unstable(); sorted.dedup(); sorted.hash(&mut hash); } + config.append.rollout_append_local.hash(&mut hash); hash.finish() } fn enabled(config: &MasterConfig) -> bool { - config.append.rollout_append_local + crate::local_merge::enabled(config) && config.append.rollout_append_reconcile_interval_secs > 0 && config.merge_wal_interval_secs > 0 } @@ -53,13 +55,12 @@ fn candidates(config: &MasterConfig, snapshot: &[StatRow], after: Option<&str>) .owned_targets .iter() .chain(config.append.rollout_append_targets.iter()) + .chain(config.local_merge.master_local_merge_targets.iter()) .map(String::as_str) .chain(snapshot.iter().map(|row| row.name.as_str())) .filter(|name| { *name != "*" - && config.append.local(name) - && config.merge_rollout.owned(name) - && !config.merge_rollout.draining(name) + && crate::local_merge::selected(config, name) && after.is_none_or(|after| *name > after) }) .map(str::to_owned) @@ -236,20 +237,19 @@ async fn probe(state: &Arc, target: &str) -> Result { { return Ok(false); } - let coordinator = AppendCoordinator::open( - &state.rollout_uri(target), - // Reuse this master's small local-staging session, never default caches. + let uri = target + .strip_prefix("generic:") + .map_or_else(|| state.rollout_uri(target), |name| state.generic_uri(name)); + let pending = has_sealed_wal( + &uri, state .local_append .as_ref() .map(|local| local.session.clone()), + &state.config.catchup.shards, ) .await .map_err(|e| e.to_string())?; - let pending = coordinator - .has_sealed_wal(&state.config.catchup.shards) - .await - .map_err(|e| e.to_string())?; if pending { state .task_store @@ -482,6 +482,206 @@ mod tests { assert!(!probe(&successor, "a").await.unwrap()); } + async fn generic_fixture() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().unwrap(); + let mut config = MasterConfig::parse_from(["test"]); + config.data_dir = dir.path().to_string_lossy().into(); + config.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + config.etcd.etcd_prefix = format!( + "/resident-generic-test/{}", + lance_context_core::generate_id() + ); + config.merge_rollout.owned_targets = vec!["generic:a".into(), "generic:b".into()]; + config.local_merge.master_local_merge_targets = vec!["*".into()]; + config.catchup.shards = vec!["writer".into()]; + config.catchup.merge_max_generations = 1; + config.catchup.merge_max_bytes = 1024 * 1024; + config.append.rollout_append_local_memory_bytes = 2 * 1024 * 1024; + config.worker_endpoints.clear(); + config.stats_scan_interval_secs = 0; + (dir, MasterState::new(config).await.unwrap()) + } + + async fn generic_writer(state: &MasterState, name: &str) -> lance_context_core::GenericStore { + use lance_context_core::{ + ColumnSpec, ColumnType, GenericStore, GenericStoreOptions, SchemaSpec, + }; + GenericStore::open( + &state.generic_uri(name), + SchemaSpec::new(vec![ + ( + "id".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ( + "text".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ]), + GenericStoreOptions { + shard_id: Some("writer".into()), + merge_after_generations: Some(0), + ..Default::default() + }, + ) + .await + .unwrap() + } + + async fn generic_write(writer: &lance_context_core::GenericStore, id: &str, text: &str) { + writer + .add(&[serde_json::json!({"id":id, "text":text}) + .as_object() + .unwrap() + .clone()]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn generic_discovery_and_continuation_preserve_updates_without_workers() { + let (_dir, state) = generic_fixture().await; + let writer = generic_writer(&state, "a").await; + for n in 0..17 { + generic_write(&writer, "shared", &n.to_string()).await; + } + assert_eq!(writer.pending_wal_generations().await.unwrap(), 17); + // A schema-independent manifest probe discovers generic demand without + // payload reads or a stats scan. A second master cannot duplicate it. + let second = MasterState::new(state.config.clone()).await.unwrap(); + let now = chrono::Utc::now().timestamp_millis(); + let (a, b) = tokio::join!(tick(&state, now), tick(&second, now)); + assert_eq!(a.unwrap() + b.unwrap(), 1); + let coordinator = state.task_store.merge_coordinator(); + let first = coordinator.request_page(None).await.unwrap().0.remove(0); + crate::scheduler::enqueue_merge_request(&state, &coordinator, &first) + .await + .unwrap(); + let claim = state + .task_store + .claim_resident_merge(&crate::local_merge::targets(&state.config)) + .await + .unwrap() + .unwrap(); + assert!(second + .task_store + .claim_resident_merge(&crate::local_merge::targets(&second.config)) + .await + .unwrap() + .is_none()); + let result = crate::merge_execution::run_merge_wal(&state, &claim).await; + assert!(result.as_ref().unwrap().contains("16 generations")); + let next = coordinator.request_page(None).await.unwrap().0.remove(0); + crate::scheduler::enqueue_merge_request(&second, &coordinator, &next) + .await + .unwrap(); + assert_eq!(coordinator.request_page(None).await.unwrap().0.len(), 1); + state.task_store.finish(claim, result).await.unwrap(); + crate::scheduler::enqueue_merge_request(&second, &coordinator, &next) + .await + .unwrap(); + let claim = second + .task_store + .claim_resident_merge(&crate::local_merge::targets(&second.config)) + .await + .unwrap() + .unwrap(); + let result = crate::merge_execution::run_merge_wal(&second, &claim).await; + assert!(result.as_ref().unwrap().contains("1 generations")); + second.task_store.finish(claim, result).await.unwrap(); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 0); + let rows = writer.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["text"], "16"); + assert!(!probe(&second, "generic:a").await.unwrap()); + assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0); + assert_eq!(second.local_append.as_ref().unwrap().budget.reserved(), 0); + // No ingestion epoch was acquired by either master. + generic_write(&writer, "shared", "after").await; + assert!(probe(&second, "generic:a").await.unwrap()); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn generic_replicas_merge_distinct_tables_during_live_ingestion() { + let (_dir, state) = generic_fixture().await; + let a = generic_writer(&state, "a").await; + let b = generic_writer(&state, "b").await; + for n in 0..4 { + generic_write(&a, "shared", &n.to_string()).await; + generic_write(&b, "shared", &n.to_string()).await; + } + for target in ["generic:a", "generic:b"] { + state + .task_store + .enqueue(TaskKind::MergeWal, target, vec![]) + .await + .unwrap(); + } + let second = MasterState::new(state.config.clone()).await.unwrap(); + let targets = crate::local_merge::targets(&state.config); + let claim_a = state + .task_store + .claim_resident_merge(&targets) + .await + .unwrap() + .unwrap(); + let claim_b = second + .task_store + .claim_resident_merge(&targets) + .await + .unwrap() + .unwrap(); + assert_ne!(claim_a.task.target, claim_b.task.target); + let budget = state.local_append.as_ref().unwrap().budget.clone(); + let held = budget.reserve(budget.limit()).await; + let runner = { + let state = state.clone(); + tokio::spawn(async move { + let result = crate::merge_execution::run_merge_wal(&state, &claim_a).await; + assert!(result.is_ok(), "{result:?}"); + state.task_store.finish(claim_a, result).await.unwrap(); + }) + }; + let (rb, ()) = tokio::join!( + crate::merge_execution::run_merge_wal(&second, &claim_b), + async { + for n in 0..8 { + generic_write(&a, &format!("during-{n}"), "live").await; + generic_write(&b, &format!("during-{n}"), "live").await; + } + } + ); + assert!(rb.is_ok(), "{rb:?}"); + second.task_store.finish(claim_b, rb).await.unwrap(); + assert!( + !runner.is_finished(), + "local generic merge must respect the process budget" + ); + drop(held); + tokio::time::timeout(Duration::from_secs(30), runner) + .await + .unwrap() + .unwrap(); + for writer in [&a, &b] { + generic_write(writer, "after", "live").await; + let rows = writer.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 10); + assert_eq!( + rows.iter().find(|row| row["id"] == "shared").unwrap()["text"], + "3" + ); + } + assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0); + assert_eq!(second.local_append.as_ref().unwrap().budget.reserved(), 0); + } + #[test] fn explicit_owned_targets_rotate_without_any_stats_and_exclude_legacy() { let mut config = MasterConfig::parse_from(["test"]); diff --git a/crates/lance-context-master/src/rollout_append.rs b/crates/lance-context-master/src/rollout_append.rs index c7014c59..c777e6c4 100644 --- a/crates/lance-context-master/src/rollout_append.rs +++ b/crates/lance-context-master/src/rollout_append.rs @@ -114,7 +114,7 @@ impl AppendConfig { .iter() .any(|t| t == "*" || t == target) } - fn validate(&self) -> Result<()> { + pub(crate) fn validate(&self) -> Result<()> { if !(1..=32).contains(&self.rollout_append_concurrency) || !(1..=256).contains(&self.rollout_append_max_generations) || self.rollout_append_max_bytes == 0 @@ -222,7 +222,8 @@ pub(crate) async fn run(state: &Arc, target: &str) -> Result = state .config .worker_endpoints @@ -309,7 +310,8 @@ pub(crate) async fn run(state: &Arc, target: &str) -> Result MasterConfig { MasterConfig { + local_merge: Default::default(), append: Default::default(), catchup: Default::default(), wal_tail: Default::default(), diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 6b1c0f0c..e787731b 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -1142,7 +1142,7 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { if let Some(merge) = merge_sem { spawn_pool_poller(state.clone(), merge, TaskKinds::MERGE_WAL, false); } - if state.config.append.rollout_append_local + if crate::local_merge::enabled(&state.config) && state.config.append.rollout_append_local_task_concurrency > 0 { // Reserve before claiming. A saturated legacy RPC pool cannot consume @@ -1239,7 +1239,7 @@ fn spawn_filtered_pool_poller( let claim = if resident_only { state .task_store - .claim_resident_merge(&state.config.append.rollout_append_targets) + .claim_resident_merge(&crate::local_merge::targets(&state.config)) .await } else { state.task_store.claim_next_of_kinds(kinds).await @@ -1291,6 +1291,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + local_merge: Default::default(), append: Default::default(), catchup: Default::default(), wal_tail: Default::default(), @@ -1896,7 +1897,9 @@ mod tests { let mut cfg = config(&dir); cfg.merge_rollout.owned_targets.retain(|t| t != "exp"); cfg.merge_rollout.drain_targets.push("exp".into()); - cfg.worker_endpoints = vec!["http://127.0.0.1:1".into()]; + // A resident-only master must reconcile the recorded old execution + // even though it has no configured worker endpoints for fresh work. + cfg.worker_endpoints.clear(); let state = MasterState::new(cfg).await.unwrap(); enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap(); let claim = state diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index e4d0c12f..1dd98722 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -138,6 +138,7 @@ impl MasterState { pub async fn new(config: MasterConfig) -> lance::Result> { config.merge_rollout.validate().map_err(lance::Error::io)?; config.catchup.validate().map_err(lance::Error::io)?; + crate::local_merge::validate(&config).map_err(lance::Error::io)?; config .append .validate_local(&config) @@ -216,7 +217,7 @@ impl MasterState { if let Some(guard) = init_guard { task_store.release_coordination_lock(guard).await?; } - let local_append = config.append.rollout_append_local.then(|| { + let local_append = crate::local_merge::enabled(&config).then(|| { crate::rollout_append::LocalStaging::new( config.append.rollout_append_local_memory_bytes, ) @@ -319,6 +320,7 @@ mod tests { fn test_config(dir: &TempDir) -> MasterConfig { MasterConfig { + local_merge: Default::default(), append: Default::default(), catchup: Default::default(), wal_tail: Default::default(), diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 38794a94..339eeeed 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -2336,6 +2336,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + local_merge: Default::default(), append: Default::default(), catchup: Default::default(), wal_tail: Default::default(), @@ -2576,7 +2577,7 @@ mod tests { .put(key.clone(), "external-owner", None) .await .unwrap(); - let targets = ["hot", "draining", "external", "generic:gs"].map(str::to_owned); + let targets = ["hot", "draining", "external"].map(str::to_owned); let claim = store.claim_resident_merge(&targets).await.unwrap().unwrap(); assert_eq!(claim.task.target, "hot"); // A second replica cannot take the running canonical task. @@ -2609,6 +2610,18 @@ mod tests { .await .unwrap() .is_none()); + let generic = store + .claim_resident_merge(&["generic:gs".into()]) + .await + .unwrap() + .unwrap(); + assert_eq!(generic.task.target, "generic:gs"); + assert!(other + .claim_resident_merge(&["generic:gs".into()]) + .await + .unwrap() + .is_none()); + store.finish(generic, Ok("done".into())).await.unwrap(); store .inner .client diff --git a/crates/lance-context-master/src/wal_tail.rs b/crates/lance-context-master/src/wal_tail.rs index 2cce2b5b..76888c5e 100644 --- a/crates/lance-context-master/src/wal_tail.rs +++ b/crates/lance-context-master/src/wal_tail.rs @@ -38,7 +38,7 @@ struct Cursor { fn enabled(config: &MasterConfig) -> bool { config.wal_tail.wal_tail_interval_secs > 0 && config.merge_wal_interval_secs > 0 - && (!config.worker_endpoints.is_empty() || config.append.rollout_append_local) + && (!config.worker_endpoints.is_empty() || crate::local_merge::enabled(config)) && config.wal_tail.wal_tail_stats_max_age_secs > 0 } @@ -58,7 +58,7 @@ fn candidates( && now.saturating_sub(row.scanned_at) <= age && !config.merge_rollout.draining(&row.name) && (!config.worker_endpoints.is_empty() - || (config.append.local(&row.name) && config.merge_rollout.owned(&row.name))) + || crate::local_merge::selected(config, &row.name)) && after.is_none_or(|after| row.name.as_str() > after) }) .map(|row| row.name.clone()) diff --git a/docs/resident-wal-recovery.md b/docs/resident-wal-recovery.md index 9f6bf7b3..13733379 100644 --- a/docs/resident-wal-recovery.md +++ b/docs/resident-wal-recovery.md @@ -1,6 +1,6 @@ # Autonomous resident WAL demand -Resident masters already execute bounded append-only merges using the ordinary +Resident masters execute bounded WAL merges using the ordinary durable queue, shared process memory budget, progress watchdog and publication fences. Two scheduling gaps could still require an external enqueue loop: @@ -16,7 +16,8 @@ and existing maintenance admission; a hot table does not retain its claim indefinitely. If the last pass drained everything, at most one empty successor is needed; a zero-work pass does not perpetuate itself. -With `ROLLOUT_APPEND_LOCAL=true`, `MERGE_WAL_INTERVAL_SECS>0`, and +With `ROLLOUT_APPEND_LOCAL=true` or a nonempty `MASTER_LOCAL_MERGE_TARGETS`, +`MERGE_WAL_INTERVAL_SECS>0`, and `ROLLOUT_APPEND_RECONCILE_INTERVAL_SECS>0` (default 30), masters also discover WAL demand separately from the stats scanner. One fleet-wide etcd compare-and-swap reserves four targets per batch, with two metadata readers, a ten-second limit @@ -53,7 +54,7 @@ must retain their existing recovery protection until separately validated. Resident masters also reserve `ROLLOUT_APPEND_LOCAL_TASK_CONCURRENCY` task slots (default 1, 0 disables the reservation) in addition to `MERGE_WAL_CONCURRENCY`. -Only locally enabled, owned, non-draining rollout merges can use these slots. +Only locally enabled, owned, non-draining merges can use these slots. Eligibility is checked before dependency reads or claim/lock mutation. The ordinary merge pool remains available to all merge tasks, including legacy worker RPCs; a legacy request cannot consume the reserved capacity. Local @@ -61,3 +62,34 @@ Arrow buffers still share `ROLLOUT_APPEND_LOCAL_MEMORY_BYTES` across both pools. Canonical claims, external catch-up ownership, progress watchdogs, and process admission drain apply equally to the reserved pool. This does not cancel an existing legacy owner of the same table. + +## Generic and ordinary rollout tables + +`MASTER_LOCAL_MERGE_TARGETS=table,generic:rows` executes the selected tables in +native master processes. `*` selects the explicitly owned, non-draining targets +in `MERGE_OWNED_TARGETS`; it does not enroll legacy tables into ownership. +Existing `ROLLOUT_APPEND_LOCAL` selection continues to select rollouts only. + +Append-opted-in tables keep parallel append file preparation. Other rollouts +and generic tables use the existing bounded, ordered catch-up implementation, +including upserts and key-index maintenance. All local work shares the same +`ROLLOUT_APPEND_LOCAL_MEMORY_BYTES` budget and small metadata session. The +ordinary path uses `CATCHUP_MERGE_MAX_BYTES`, `CATCHUP_MERGE_MAX_GENERATIONS`, +`CATCHUP_PIPELINE_ENABLED`, and `CATCHUP_SLICE_SECS`. Each batch must fit the +configured shared budget; an indivisible generation can exceed the batch byte +cap, subject to the core reservation rules. This budget bounds merge Arrow +reservations, not total process RSS or all encoding/index scratch allocations. + +Both paths publish continuation demand after a bounded slice. Metadata discovery +supports generic schemas without opening ingestion writers. Stable shard names +must include every writer to be serviced. No Kubernetes Pod creation or pool +inventory permission is required. To add execution capacity, scale the master +replicas; canonical claims distribute different tables, while the same table +retains one fenced publisher. Every replica has its own bounded buffer budget. + +Activation still requires ownership migration: drain the target's old unfenced +maintenance writers, prove their admitted storage writes finished, then enable +owned maintenance consistently across writers, masters and helpers. Keep +existing consumers running for other tables until their own handoff is complete. +This flag neither imports private recovery ledgers nor turns unknown old +executions into safe takeover candidates.