feat(master): auto-sweep MergeWal from pending WAL generations

crates/lance-context-master/src/config.rs

@@ -40,6 +40,19 @@ pub struct MasterConfig {
     #[arg(long, env = "TARGET_ROWS_PER_FRAGMENT", default_value_t = 1_048_576)]
     pub target_rows_per_fragment: usize,
 
+    /// Interval, in seconds, between automatic WAL-merge sweeps. Each sweep
+    /// enqueues a `MergeWal` task for every experiment whose pending MemWAL
+    /// generation count crosses `merge_wal_min_generations`; the task fans out
+    /// to the configured worker endpoints. `0` disables automatic WAL merge
+    /// (manual triggers still work).
+    #[arg(long, env = "MERGE_WAL_INTERVAL_SECS", default_value_t = 600)]
+    pub merge_wal_interval_secs: u64,
+
+    /// Minimum pending MemWAL generations (per experiment) before an automatic
+    /// `MergeWal` is enqueued. Read from the periodically-scanned stats table.
+    #[arg(long, env = "MERGE_WAL_MIN_GENERATIONS", default_value_t = 8)]
+    pub merge_wal_min_generations: i64,
+
     /// Data-plane worker base URLs (comma-separated), e.g.
     /// `http://rollout-0:3000,http://rollout-1:3000`. A `MergeWal` task fans out
     /// to every endpoint so each worker merges its own MemWAL shard (the master

crates/lance-context-master/src/routes.rs

@@ -438,6 +438,8 @@ mod tests {
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,
+            merge_wal_interval_secs: 0,
+            merge_wal_min_generations: 8,
             worker_endpoints: vec![],
             task_concurrency: 4,
             etcd_endpoints: test_etcd_endpoints(),

crates/lance-context-master/src/scheduler.rs

@@ -54,11 +54,12 @@ fn kind_label(kind: TaskKind) -> &'static str {
     }
 }
 
-/// Enqueue a task and return its record. For [`TaskKind::Compact`] and
-/// [`TaskKind::IndexId`] this de-dupes against an existing non-terminal task for
-/// the same target: if one is already `Queued` or `Running`, its record is
-/// returned unchanged and nothing new is enqueued. `MergeWal` tasks are not
-/// de-duped (each fan-out is independent).
+/// Enqueue a task and return its record. For [`TaskKind::Compact`],
+/// [`TaskKind::IndexId`], and depless [`TaskKind::MergeWal`] this de-dupes
+/// against an existing non-terminal task for the same target: if one is already
+/// `Queued` or `Running`, its record is returned unchanged and nothing new is
+/// enqueued. A `MergeWal` that is part of a dependency chain (non-empty
+/// `depends_on`) is not de-duped.
 pub async fn enqueue(
     state: &Arc<MasterState>,
     kind: TaskKind,
@@ -305,9 +306,44 @@ fn in_quiet_hours(config: &CompactionConfig) -> bool {
         .any(|(start, end)| hour >= *start && hour < *end)
 }
 
+/// Enqueue a `MergeWal` task for every experiment whose pending MemWAL
+/// generation count is at or above the configured threshold, reading candidates
+/// from the stats table. Coordinated across master replicas by a dedicated
+/// task-store lock so only one replica sweeps at a time. Depless `MergeWal`
+/// enqueues de-dupe, so a still-running fan-out is not re-queued.
+pub async fn sweep_merge_wal_candidates(state: &Arc<MasterState>) -> lance::Result<usize> {
+    let Some(guard) = state
+        .task_store
+        .try_coordination_lock("merge-wal-sweep")
+        .await?
+    else {
+        return Ok(0);
+    };
+    let result = sweep_merge_wal_inner(state).await;
+    let release = state.task_store.release_coordination_lock(guard).await;
+    match (result, release) {
+        (Ok(count), Ok(())) => Ok(count),
+        (Err(error), _) => Err(error),
+        (Ok(_), Err(error)) => Err(error),
+    }
+}
+
+async fn sweep_merge_wal_inner(state: &Arc<MasterState>) -> lance::Result<usize> {
+    let threshold = state.config.merge_wal_min_generations;
+    let rows = state.stats.lock().await.list(None, usize::MAX, 0).await?;
+    let mut queued = 0;
+    for row in rows {
+        if row.pending_wal_generations >= threshold {
+            enqueue(state, TaskKind::MergeWal, &row.name).await?;
+            queued += 1;
+        }
+    }
+    Ok(queued)
+}
+
 /// Spawn the scheduler poller plus the optional periodic auto-sweep.
 pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
-    // Optional periodic auto-sweep feeds the same queue.
+    // Optional periodic compaction auto-sweep feeds the same queue.
     let interval_secs = state.config.compaction_interval_secs;
     if interval_secs > 0 {
         let sweep_state = state.clone();
@@ -325,6 +361,26 @@ pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
         });
     }
 
+    // Optional periodic WAL-merge auto-sweep, independent of compaction.
+    let merge_interval_secs = state.config.merge_wal_interval_secs;
+    if merge_interval_secs > 0 {
+        let sweep_state = state.clone();
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(Duration::from_secs(merge_interval_secs));
+            ticker.tick().await; // skip immediate tick
+            loop {
+                ticker.tick().await;
+                match sweep_merge_wal_candidates(&sweep_state).await {
+                    Ok(n) if n > 0 => {
+                        tracing::info!(queued = n, "auto merge-wal sweep queued experiments")
+                    }
+                    Ok(_) => {}
+                    Err(e) => tracing::warn!(error = %e, "auto merge-wal sweep failed"),
+                }
+            }
+        });
+    }
+
     let concurrency = state.config.task_concurrency.max(1);
     let sem = Arc::new(Semaphore::new(concurrency));
     let dispatch_state = state.clone();
@@ -378,6 +434,8 @@ mod tests {
             // Low threshold so a handful of appends crosses it.
             min_fragments: 2,
             target_rows_per_fragment: 1_048_576,
+            merge_wal_interval_secs: 0,
+            merge_wal_min_generations: 2,
             worker_endpoints: vec![],
             task_concurrency: 4,
             etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS")
@@ -636,6 +694,60 @@ mod tests {
         worker.abort();
     }
 
+    /// The WAL-merge sweep enqueues a `MergeWal` only for experiments whose
+    /// pending generation count is at or above the threshold, and de-dupes so a
+    /// second sweep does not pile up a duplicate for the same target.
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn sweep_merge_wal_enqueues_over_threshold_and_dedupes() {
+        use crate::stats_store::StatRow;
+
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.merge_wal_min_generations = 3;
+        let state = MasterState::new(cfg).await.unwrap();
+
+        let seed = |name: &str, pending: i64| StatRow {
+            name: name.to_string(),
+            uri: state.rollout_uri(name),
+            row_count: 0,
+            fragment_count: 0,
+            last_updated: 0,
+            pending_wal_generations: pending,
+            last_compaction: StatRow::NO_COMPACTION,
+            total_compactions: 0,
+            scanned_at: 0,
+        };
+        {
+            let mut stats = state.stats.lock().await;
+            stats.upsert(&seed("hot", 5)).await.unwrap(); // >= threshold
+            stats.upsert(&seed("cold", 1)).await.unwrap(); // < threshold
+        }
+
+        let queued = sweep_merge_wal_candidates(&state).await.unwrap();
+        assert_eq!(queued, 1, "only the over-threshold experiment is swept");
+
+        let tasks = state.task_store.list().await.unwrap();
+        let merge_tasks: Vec<_> = tasks
+            .iter()
+            .filter(|t| t.kind == TaskKind::MergeWal)
+            .collect();
+        assert_eq!(merge_tasks.len(), 1);
+        assert_eq!(merge_tasks[0].target, "hot");
+
+        // Second sweep must de-dupe against the still-queued MergeWal.
+        sweep_merge_wal_candidates(&state).await.unwrap();
+        let merge_after = state
+            .task_store
+            .list()
+            .await
+            .unwrap()
+            .into_iter()
+            .filter(|t| t.kind == TaskKind::MergeWal)
+            .count();
+        assert_eq!(merge_after, 1, "duplicate MergeWal is de-duped");
+    }
+
     /// Minimal rollout record builder for tests (the core struct has no    /// `Default`).
     pub fn rollout_record(id: &str) -> lance_context_core::RolloutRecord {
         use chrono::TimeZone;

crates/lance-context-master/src/state.rs

@@ -93,6 +93,8 @@ mod tests {
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,
+            merge_wal_interval_secs: 0,
+            merge_wal_min_generations: 8,
             worker_endpoints: vec![],
             task_concurrency: 4,
             etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS")

crates/lance-context-master/src/task_store.rs

@@ -86,9 +86,10 @@ impl TaskStore {
         Ok(store)
     }
 
-    /// Atomically enqueue a task. Standalone Compact and IndexId tasks use an
-    /// etcd dedupe key while queued or running. Tasks with dependencies are
-    /// always distinct because they belong to a specific ordered chain.
+    /// Atomically enqueue a task. Standalone Compact, IndexId, and depless
+    /// MergeWal tasks use an etcd dedupe key while queued or running. Tasks with
+    /// dependencies are always distinct because they belong to a specific
+    /// ordered chain.
     pub async fn enqueue(
         &self,
         kind: TaskKind,
@@ -806,7 +807,15 @@ fn fail_dependency(task: &mut TaskRecord, dependency: &str) {
 }
 
 fn should_dedupe(kind: TaskKind, depends_on: &[String]) -> bool {
-    depends_on.is_empty() && matches!(kind, TaskKind::Compact | TaskKind::IndexId)
+    // MergeWal is deduped for depless enqueues (periodic auto-sweep and the
+    // manual "Merge WAL" button) so a slow fan-out cannot pile up duplicate
+    // broadcasts for the same experiment. A MergeWal that is part of an ordered
+    // Optimize chain carries `depends_on` and is intentionally not deduped.
+    depends_on.is_empty()
+        && matches!(
+            kind,
+            TaskKind::Compact | TaskKind::IndexId | TaskKind::MergeWal
+        )
 }
 
 fn requires_target_lock(kind: TaskKind) -> bool {
@@ -874,6 +883,8 @@ mod tests {
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,
+            merge_wal_interval_secs: 0,
+            merge_wal_min_generations: 8,
             worker_endpoints: vec![],
             task_concurrency: 4,
             etcd_endpoints: vec![],

deploy/kubernetes/README.md

@@ -30,3 +30,17 @@ remain idempotent across crash recovery.
 The `_stats.rollout.lance` table remains in `DATA_DIR`. etcd coordinates its
 single-writer sections, while readers reload the latest Lance manifest so every
 master replica sees current stats.
+
+## Automatic maintenance
+
+Two periodic sweeps run on the master and feed the shared scheduler queue:
+
+- **Compaction** (`COMPACTION_INTERVAL_SECS`, `MIN_FRAGMENTS`) rewrites an
+  experiment's base-table fragments locally on the master.
+- **WAL merge** (`MERGE_WAL_INTERVAL_SECS`, `MERGE_WAL_MIN_GENERATIONS`) enqueues
+  a `MergeWal` task for every experiment whose pending MemWAL generation count
+  (from the periodically-scanned stats table) crosses the threshold. The task
+  fans out to every `WORKER_ENDPOINTS` worker, each of which folds its own shard.
+  The master cannot merge a shard it does not own without fencing the live
+  writer, so the merge itself always runs on the owning worker. Set the interval
+  to `0` to disable it; the manual "Merge WAL" / "Optimize" UI actions still work.

deploy/kubernetes/master.yaml

@@ -60,6 +60,13 @@ spec:
               value: /etc/lance-context/etcd/tls.key
             - name: TASK_HISTORY_LIMIT
               value: "1000"
+            # Master periodically fans a WAL merge out to every worker for any
+            # experiment with at least this many pending MemWAL generations.
+            # Set MERGE_WAL_INTERVAL_SECS to 0 to disable auto WAL merge.
+            - name: MERGE_WAL_INTERVAL_SECS
+              value: "600"
+            - name: MERGE_WAL_MIN_GENERATIONS
+              value: "8"
             - name: MASTER_PORT
               value: "8090"
             - name: UI_DIR