diff --git a/crates/lance-context-core/src/lib.rs b/crates/lance-context-core/src/lib.rs
index 7ccfaa2..4435f3e 100644
--- a/crates/lance-context-core/src/lib.rs
+++ b/crates/lance-context-core/src/lib.rs
@@ -102,3 +102,5 @@ pub use lance::Error as LanceError;
 // Re-export the Lance session type so the server can build one shared,
 // capacity-bounded cache session across all resident rollout stores.
 pub use lance::session::Session;
+
+pub use store_base::PreparedCompaction;
diff --git a/crates/lance-context-core/src/rollout_append.rs b/crates/lance-context-core/src/rollout_append.rs
index 4c5c816..13b02ca 100644
--- a/crates/lance-context-core/src/rollout_append.rs
+++ b/crates/lance-context-core/src/rollout_append.rs
@@ -934,6 +934,166 @@ mod tests {
         MergeMemoryBudget::new(8 * 1024 * 1024)
     }
 
+    #[derive(Debug)]
+    struct NoPreparationCommit;
+    impl crate::merge_write_scope::CommitAuthorizer for NoPreparationCommit {
+        fn authorize<'a>(
+            &'a self,
+            _: &'a str,
+            _: u64,
+        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = lance::Result<()>> + Send + 'a>>
+        {
+            Box::pin(async { Err(Error::io("unexpected preparation manifest commit")) })
+        }
+    }
+
+    async fn prepared_compaction_fixture(uri: &str) -> (RolloutStore, AppendCoordinator) {
+        let a = writer(uri, "a").await;
+        let mut coordinator = AppendCoordinator::open(uri, None).await.unwrap();
+        for i in 0..4 {
+            put(&a, &format!("seed-{i}"), 4096).await;
+            let (plans, _) = coordinator
+                .plan(&["a".into()], 64, 1024 * 1024)
+                .await
+                .unwrap();
+            let part = stage(uri, plans[0].clone(), budget(), None).await.unwrap();
+            coordinator.commit(vec![part]).await.unwrap();
+        }
+        (a, coordinator)
+    }
+
+    async fn prepare_without_commit(uri: &str) -> crate::PreparedCompaction {
+        crate::merge_write_scope::MergeWriteScope::with_pinned_authorizer(Arc::new(
+            NoPreparationCommit,
+        ))
+        .run(async {
+            let store =
+                RolloutStore::open_existing_with_options(uri, RolloutStoreOptions::default())
+                    .await
+                    .unwrap();
+            store
+                .prepare_compaction(crate::CompactionConfig {
+                    target_rows_per_fragment: 1024,
+                    max_source_fragments: Some(128),
+                    ..Default::default()
+                })
+                .await
+                .unwrap()
+        })
+        .await
+    }
+
+    #[tokio::test]
+    async fn prepared_compaction_preserves_append_and_watermarks_in_both_orders() {
+        for merge_first in [true, false] {
+            let dir = tempfile::tempdir().unwrap();
+            let uri = dir.path().to_str().unwrap();
+            let (a, mut coordinator) = prepared_compaction_fixture(uri).await;
+            let before = Dataset::open(uri).await.unwrap().version().version;
+            let prepared = prepare_without_commit(uri).await;
+            assert!(!prepared.is_empty());
+            assert_eq!(
+                Dataset::open(uri).await.unwrap().version().version,
+                before,
+                "preparation must not publish even a reservation"
+            );
+            put(&a, "during-compact", 8192).await;
+            let (plans, _) = coordinator
+                .plan(&["a".into()], 64, 1024 * 1024)
+                .await
+                .unwrap();
+            let staged = stage(uri, plans[0].clone(), budget(), None).await.unwrap();
+            let high = staged.plan.generations.last().unwrap().number;
+            let shard = staged.plan.shard;
+            let mut publisher =
+                RolloutStore::open_existing_with_options(uri, RolloutStoreOptions::default())
+                    .await
+                    .unwrap();
+            if merge_first {
+                coordinator.commit(vec![staged.clone()]).await.unwrap();
+            }
+            let metrics = publisher
+                .commit_prepared_compaction(prepared)
+                .await
+                .unwrap();
+            assert_eq!(metrics.fragments_removed, 4);
+            if !merge_first {
+                coordinator.commit(vec![staged.clone()]).await.unwrap();
+            }
+            assert_eq!(
+                coordinator.commit(vec![staged]).await.unwrap(),
+                0,
+                "replay after rewrite must not duplicate rows"
+            );
+            assert_eq!(rows(uri).await, 5);
+            let current = Dataset::open(uri).await.unwrap();
+            assert_eq!(watermarks(&current).await.unwrap().get(&shard), Some(&high));
+            let reader =
+                RolloutStore::open_existing_with_options(uri, RolloutStoreOptions::default())
+                    .await
+                    .unwrap();
+            for i in 0..4 {
+                assert_eq!(
+                    reader.get_blob(&format!("seed-{i}")).await.unwrap(),
+                    Some(vec![42; 4096])
+                );
+            }
+            assert_eq!(
+                reader.get_blob("during-compact").await.unwrap(),
+                Some(vec![42; 8192])
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn prepared_compaction_cannot_publish_with_revoked_authority() {
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let (_writer, _) = prepared_compaction_fixture(uri).await;
+        let prepared = prepare_without_commit(uri).await;
+        let before = Dataset::open(uri).await.unwrap().version().version;
+        let result = crate::merge_write_scope::MergeWriteScope::with_pinned_authorizer(Arc::new(
+            NoPreparationCommit,
+        ))
+        .run(async {
+            let mut publisher =
+                RolloutStore::open_existing_with_options(uri, RolloutStoreOptions::default())
+                    .await
+                    .unwrap();
+            publisher.commit_prepared_compaction(prepared).await
+        })
+        .await;
+        assert!(result.is_err());
+        assert_eq!(Dataset::open(uri).await.unwrap().version().version, before);
+        assert_eq!(rows(uri).await, 4);
+    }
+
+    #[tokio::test]
+    async fn prepared_compaction_rejects_deleted_sources_without_resurrection() {
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let (_a, _) = prepared_compaction_fixture(uri).await;
+        let prepared = prepare_without_commit(uri).await;
+        let mut latest = Dataset::open(uri).await.unwrap();
+        latest.delete("id = 'seed-0'").await.unwrap();
+        let version = latest.version().version;
+        let mut publisher =
+            RolloutStore::open_existing_with_options(uri, RolloutStoreOptions::default())
+                .await
+                .unwrap();
+        let error = publisher
+            .commit_prepared_compaction(prepared)
+            .await
+            .unwrap_err();
+        assert!(
+            error.to_string().contains("source fragment changed"),
+            "{error}"
+        );
+        assert_eq!(Dataset::open(uri).await.unwrap().version().version, version);
+        assert_eq!(rows(uri).await, 3);
+        assert!(publisher.get_blob("seed-0").await.unwrap().is_none());
+    }
+
     #[tokio::test]
     async fn parallel_files_publish_once_and_preserve_new_writes() {
         let dir = tempfile::tempdir().unwrap();
diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs
index 9d47a00..712a589 100644
--- a/crates/lance-context-core/src/rollout_store.rs
+++ b/crates/lance-context-core/src/rollout_store.rs
@@ -706,6 +706,22 @@ impl RolloutStore {
         self.base.compact(options).await
     }
 
+    /// Prepare immutable compaction output without a base-table commit.
+    pub async fn prepare_compaction(
+        &self,
+        options: CompactionConfig,
+    ) -> LanceResult<crate::PreparedCompaction> {
+        self.base.prepare_compaction(options).await
+    }
+
+    /// Publish prepared output using this freshly opened, fenced handle.
+    pub async fn commit_prepared_compaction(
+        &mut self,
+        prepared: crate::PreparedCompaction,
+    ) -> LanceResult<CompactionMetrics> {
+        self.base.commit_prepared_compaction(prepared).await
+    }
+
     /// See `StorageBase::base_data_files`.
     pub fn base_data_files(&self) -> Vec<String> {
         self.base.base_data_files()
diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index 3f6fbd5..8bd2ffa 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -51,7 +51,7 @@ use lance::dataset::mem_wal::{
 };
 use lance::dataset::optimize::{
     commit_compaction, compact_files, plan_compaction, CompactionMetrics, CompactionMode,
-    CompactionOptions,
+    CompactionOptions, RewriteResult,
 };
 use lance::dataset::{
     builder::DatasetBuilder, Dataset, MergeInsertBuilder, NewColumnTransform, WhenMatched,
@@ -111,17 +111,38 @@ pub(crate) const DEFAULT_MERGE_MAX_BYTES: usize = 1024 * 1024 * 1024;
 /// read unaffordable: at ~1,500 per shard, reads exhausted a 32 GiB worker.
 pub(crate) const DEFAULT_PENDING_GENERATIONS_WARN: usize = 256;
 
+/// Unpublished rewrite files. No dataset handle or commit authorization escapes
+/// preparation. A fresh, fenced handle must publish this result.
+#[derive(Debug)]
+pub struct PreparedCompaction {
+    read_version: u64,
+    uri: String,
+    schema: Schema,
+    options: CompactionOptions,
+    tasks: Vec<RewriteResult>,
+}
+
+impl PreparedCompaction {
+    pub fn read_version(&self) -> u64 {
+        self.read_version
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.tasks.is_empty()
+    }
+}
+
 /// 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
 /// that exceeds the budget. A long run of tiny fragments can therefore produce
 /// one oversized task and compact nothing. Truncating the public task data
 /// keeps the rewrite contiguous while guaranteeing incremental progress.
-async fn compact_files_incremental(
-    dataset: &mut Dataset,
+async fn prepare_compaction_files(
+    dataset: &Dataset,
     mut options: CompactionOptions,
     max_source_fragments: usize,
-) -> LanceResult<CompactionMetrics> {
+) -> LanceResult<Vec<RewriteResult>> {
     options.max_source_fragments = None;
     let plan = plan_compaction(dataset, &options).await?;
     let mut remaining = max_source_fragments;
@@ -137,7 +158,7 @@ async fn compact_files_incremental(
         }
     }
     if tasks.is_empty() {
-        return Ok(CompactionMetrics::default());
+        return Ok(Vec::new());
     }
 
     let concurrency = options.num_threads.unwrap_or(1).max(1);
@@ -151,6 +172,16 @@ async fn compact_files_incremental(
         .try_collect()
         .await?;
 
+    Ok(completed)
+}
+
+async fn compact_files_incremental(
+    dataset: &mut Dataset,
+    options: CompactionOptions,
+    max_source_fragments: usize,
+) -> LanceResult<CompactionMetrics> {
+    let completed =
+        prepare_compaction_files(dataset, options.clone(), max_source_fragments).await?;
     commit_compaction(
         dataset,
         completed,
@@ -1527,6 +1558,97 @@ impl StorageBase {
 
     // ------------------------------------------------- compaction & indexing
 
+    fn compaction_options(config: &CompactionConfig) -> CompactionOptions {
+        CompactionOptions {
+            target_rows_per_fragment: config.target_rows_per_fragment,
+            max_rows_per_group: config.max_rows_per_group,
+            materialize_deletions: config.materialize_deletions,
+            materialize_deletions_threshold: config.materialize_deletions_threshold,
+            num_threads: config.num_threads,
+            max_bytes_per_file: config.max_bytes_per_file,
+            batch_size: config.batch_size,
+            max_source_fragments: config.max_source_fragments,
+            compaction_mode: config
+                .try_binary_copy
+                .then_some(CompactionMode::TryBinaryCopy),
+            // Every base table here carries a MemWAL index, which is fieldless
+            // (it tracks shard/generation bookkeeping, not a data column).
+            // Lance's inline index remap panics on a fieldless index ("An index
+            // existed with no fields"), so defer remapping: compaction records a
+            // fragment-reuse index and remaps lazily instead of touching the
+            // MemWAL index during the rewrite.
+            defer_index_remap: true,
+            ..Default::default()
+        }
+    }
+
+    /// Read a fixed snapshot and write new immutable files, without publishing.
+    /// Callers must forbid manifest commits while this future runs.
+    pub async fn prepare_compaction(
+        &self,
+        config: CompactionConfig,
+    ) -> LanceResult<PreparedCompaction> {
+        self.ensure_writable()?;
+        let options = Self::compaction_options(&config);
+        let tasks = prepare_compaction_files(
+            &self.dataset,
+            options.clone(),
+            config.max_source_fragments.unwrap_or(usize::MAX).max(1),
+        )
+        .await?;
+        Ok(PreparedCompaction {
+            read_version: self.dataset.version().version,
+            uri: self.dataset.uri().to_owned(),
+            schema: Schema::from(self.dataset.schema()),
+            options,
+            tasks,
+        })
+    }
+
+    /// Publish prepared files under a fresh owner. Preserve the original
+    /// RewriteResult read versions so Lance checks every intervening write.
+    pub async fn commit_prepared_compaction(
+        &mut self,
+        prepared: PreparedCompaction,
+    ) -> LanceResult<CompactionMetrics> {
+        self.ensure_writable()?;
+        self.dataset.checkout_latest().await?;
+        if prepared.uri != self.dataset.uri()
+            || prepared.schema != Schema::from(self.dataset.schema())
+        {
+            return Err(LanceError::invalid_input(
+                "compaction dataset or schema changed; reprepare",
+            ));
+        }
+        // Reject a stale rewrite before any ReserveFragments/index commit.
+        // Append-only merges leave these source fragments unchanged.
+        let sources: HashMap<_, _> = self
+            .dataset
+            .manifest()
+            .fragments
+            .iter()
+            .map(|f| (f.id, f))
+            .collect();
+        for task in &prepared.tasks {
+            for old in &task.original_fragments {
+                if sources.get(&old.id).copied() != Some(old) {
+                    return Err(LanceError::invalid_input(
+                        "compaction source fragment changed; reprepare",
+                    ));
+                }
+            }
+        }
+        let metrics = commit_compaction(
+            &mut self.dataset,
+            prepared.tasks,
+            Arc::new(DatasetIndexRemapperOptions::default()),
+            &prepared.options,
+        )
+        .await?;
+        self.reload().await?;
+        Ok(metrics)
+    }
+
     /// Compact the base table's small fragments into larger ones.
     ///
     /// Every WAL merge `append`s a new fragment to the base table, so a
@@ -1548,27 +1670,7 @@ impl StorageBase {
         self.ensure_writable()?;
         let config = options.unwrap_or_default();
 
-        let lance_options = CompactionOptions {
-            target_rows_per_fragment: config.target_rows_per_fragment,
-            max_rows_per_group: config.max_rows_per_group,
-            materialize_deletions: config.materialize_deletions,
-            materialize_deletions_threshold: config.materialize_deletions_threshold,
-            num_threads: config.num_threads,
-            max_bytes_per_file: config.max_bytes_per_file,
-            batch_size: config.batch_size,
-            max_source_fragments: config.max_source_fragments,
-            compaction_mode: config
-                .try_binary_copy
-                .then_some(CompactionMode::TryBinaryCopy),
-            // Every base table here carries a MemWAL index, which is fieldless
-            // (it tracks shard/generation bookkeeping, not a data column).
-            // Lance's inline index remap panics on a fieldless index ("An index
-            // existed with no fields"), so defer remapping: compaction records a
-            // fragment-reuse index and remaps lazily instead of touching the
-            // MemWAL index during the rewrite.
-            defer_index_remap: true,
-            ..Default::default()
-        };
+        let lance_options = Self::compaction_options(&config);
 
         let result = match config.max_source_fragments {
             Some(max_source_fragments) => {
diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs
index ad925bb..c1315a3 100644
--- a/crates/lance-context-master/src/config.rs
+++ b/crates/lance-context-master/src/config.rs
@@ -320,6 +320,15 @@ mod tests {
 /// Deadlines for local table mutations on explicitly owned targets.
 #[derive(Debug, Clone, clap::Args)]
 pub struct MaintenanceConfig {
+    /// Prepare rollout compaction files without holding the table write lock.
+    /// Explicit rollout targets or '*'; empty keeps legacy behavior. Updates to
+    /// source fragments invalidate preparation instead of being overwritten.
+    #[arg(long, env = "COMPACTION_PREPARE_TARGETS", value_delimiter = ',')]
+    pub compaction_prepare_targets: Vec<String>,
+    /// Yield prepared output after this long waiting for another table writer.
+    /// This limits local capacity retention; it never cancels the current writer.
+    #[arg(long, env = "COMPACTION_COMMIT_WAIT_SECS", default_value_t = 120)]
+    pub compaction_commit_wait_secs: u64,
     /// Legacy wire field; owned work now uses the real-progress idle timeout.
     #[arg(long, env = "MAINTENANCE_TIMEOUT_SECS", default_value_t = 3600)]
     pub maintenance_timeout_secs: u64,
@@ -333,6 +342,8 @@ pub struct MaintenanceConfig {
 impl Default for MaintenanceConfig {
     fn default() -> Self {
         Self {
+            compaction_prepare_targets: Vec::new(),
+            compaction_commit_wait_secs: 120,
             maintenance_timeout_secs: 3600,
             maintenance_idle_timeout_secs: 600,
             maintenance_drain_timeout_secs: 30,
diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs
index f909f25..c0890ab 100644
--- a/crates/lance-context-master/src/scheduler.rs
+++ b/crates/lance-context-master/src/scheduler.rs
@@ -133,7 +133,7 @@ struct TaskClaimTiming {
 /// `timing` carries how long the dispatch loop spent claiming this task and
 /// waiting for a concurrency permit, so every phase of the task's life lands on
 /// one metric rather than only the work window.
-async fn run_task(state: &Arc<MasterState>, claim: TaskClaim, timing: TaskClaimTiming) {
+async fn run_task(state: &Arc<MasterState>, mut claim: TaskClaim, timing: TaskClaimTiming) {
     let task = claim.task.clone();
     let kind = kind_label(task.kind);
 
@@ -147,28 +147,32 @@ async fn run_task(state: &Arc<MasterState>, claim: TaskClaim, timing: TaskClaimT
     .record(timing.permit_wait.as_secs_f64());
 
     let started = std::time::Instant::now();
-    let outcome = match crate::maintenance_execution::reconcile_previous(state, &claim).await {
-        Err(error) => Err(error),
-        Ok(())
-            if state.config.merge_rollout.draining(&task.target)
-                && task.kind != TaskKind::MergeWal =>
-        {
-            Err("target draining legacy writers for merge protocol transition".into())
-        }
-        Ok(()) => match task.kind {
-            TaskKind::MergeWal => crate::merge_execution::run_merge_wal(state, &claim).await,
-            _ => {
-                crate::maintenance_execution::run(state, &claim, async {
-                    match task.kind {
-                        TaskKind::Compact => run_compaction(state, &task).await,
-                        TaskKind::IndexId => run_index_id(state, &task.target).await,
-                        TaskKind::Repair => run_repair(state, &claim).await,
-                        TaskKind::MergeWal => unreachable!(),
-                    }
-                })
-                .await
+    let outcome = if claim.preparing_compaction() {
+        run_prepared_compaction(state, &mut claim).await
+    } else {
+        match crate::maintenance_execution::reconcile_previous(state, &claim).await {
+            Err(error) => Err(error),
+            Ok(())
+                if state.config.merge_rollout.draining(&task.target)
+                    && task.kind != TaskKind::MergeWal =>
+            {
+                Err("target draining legacy writers for merge protocol transition".into())
             }
-        },
+            Ok(()) => match task.kind {
+                TaskKind::MergeWal => crate::merge_execution::run_merge_wal(state, &claim).await,
+                _ => {
+                    crate::maintenance_execution::run(state, &claim, async {
+                        match task.kind {
+                            TaskKind::Compact => run_compaction(state, &task).await,
+                            TaskKind::IndexId => run_index_id(state, &task.target).await,
+                            TaskKind::Repair => run_repair(state, &claim).await,
+                            TaskKind::MergeWal => unreachable!(),
+                        }
+                    })
+                    .await
+                }
+            },
+        }
     };
     let work_elapsed = started.elapsed();
     let result = if outcome.is_ok() { "success" } else { "failed" };
@@ -320,6 +324,150 @@ async fn run_repair(state: &Arc<MasterState>, claim: &TaskClaim) -> Result<Strin
     ))
 }
 
+/// Preparation must never publish a manifest, even if a future Lance version
+/// adds a hidden metadata write. Handles opened here capture this rejection.
+#[derive(Debug)]
+struct PreparationOnly;
+impl lance_context_core::merge_write_scope::CommitAuthorizer for PreparationOnly {
+    fn authorize<'a>(
+        &'a self,
+        _: &'a str,
+        _: u64,
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = lance::Result<()>> + Send + 'a>> {
+        Box::pin(async { Err(lance::Error::io("compaction preparation cannot commit")) })
+    }
+}
+
+async fn watch_compaction_preparation(
+    state: &Arc<MasterState>,
+    claim: &TaskClaim,
+    scope: &lance_context_core::merge_write_scope::MergeWriteScope,
+) -> String {
+    let mut sequence = scope.completed_steps();
+    let mut changed = std::time::Instant::now();
+    loop {
+        tokio::time::sleep(Duration::from_secs(2)).await;
+        match state.task_store.preparation_owned(claim).await {
+            Ok(true) => {}
+            Ok(false) => return "compaction preparation claim lost".into(),
+            Err(error) => return error.to_string(),
+        }
+        let current = scope.completed_steps();
+        if current != sequence {
+            changed = std::time::Instant::now();
+            sequence = current;
+        }
+        if changed.elapsed()
+            >= Duration::from_secs(state.config.maintenance.maintenance_idle_timeout_secs)
+        {
+            return "compaction preparation made no progress".into();
+        }
+    }
+}
+
+async fn wait_for_compaction_commit(
+    state: &Arc<MasterState>,
+    claim: &mut TaskClaim,
+) -> Result<(), String> {
+    let waiting = std::time::Instant::now();
+    loop {
+        if !state
+            .task_store
+            .preparation_owned(claim)
+            .await
+            .map_err(|e| e.to_string())?
+        {
+            return Err("compaction preparation claim lost before commit".into());
+        }
+        if state
+            .task_store
+            .promote_compaction(claim)
+            .await
+            .map_err(|e| e.to_string())?
+        {
+            return Ok(());
+        }
+        // Check only after promotion has definitively declined. Never cancel
+        // an in-flight acquisition and mistake an accepted claim for a timeout.
+        if waiting.elapsed()
+            >= Duration::from_secs(state.config.maintenance.compaction_commit_wait_secs)
+        {
+            return Err("compaction commit ownership wait budget exhausted; reprepare".into());
+        }
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+}
+
+async fn run_prepared_compaction(
+    state: &Arc<MasterState>,
+    claim: &mut TaskClaim,
+) -> Result<String, String> {
+    let target = claim.task.target.clone();
+    let (_, name) = parse_target(&target);
+    let uri = state.rollout_uri(name);
+    let scope = lance_context_core::merge_write_scope::MergeWriteScope::with_pinned_authorizer(
+        Arc::new(PreparationOnly),
+    );
+    let began = std::time::Instant::now();
+    tracing::info!(task = %claim.task.id, target = %claim.task.target, "compaction file preparation started without table ownership");
+    let prepared = {
+        let work = scope.run(async {
+            let store =
+                RolloutStore::open_existing_with_options(&uri, state.rollout_store_options())
+                    .await
+                    .map_err(|e| e.to_string())?;
+            store
+                .prepare_compaction(state.compaction_config())
+                .await
+                .map_err(|e| e.to_string())
+        });
+        tokio::select! {
+            result = work => result?,
+            error = watch_compaction_preparation(state, claim, &scope) => return Err(error),
+        }
+    };
+    metrics::histogram!("master_compaction_phase_duration_seconds", "phase" => "prepare")
+        .record(began.elapsed().as_secs_f64());
+    if prepared.is_empty() {
+        // Suppress only the observed snapshot; any concurrent append changes
+        // the version and makes this hint inapplicable.
+        let options = compact_options_key(&state.compaction_config());
+        let _ = tokio::time::timeout(
+            Duration::from_secs(3),
+            state
+                .task_store
+                .record_compact_noop(name, &uri, prepared.read_version(), &options),
+        )
+        .await;
+        return Ok("compaction snapshot needs no rewrite".into());
+    }
+    let waiting = std::time::Instant::now();
+    tracing::info!(task = %claim.task.id, target = %claim.task.target, seconds = began.elapsed().as_secs_f64(), "compaction files ready; waiting for commit ownership");
+    wait_for_compaction_commit(state, claim).await?;
+    metrics::histogram!("master_compaction_phase_duration_seconds", "phase" => "commit_wait")
+        .record(waiting.elapsed().as_secs_f64());
+    // No preparation handle is reused: only this scope may authorize metadata
+    // commits, including Lance's fragment-ID reservation and index remap.
+    crate::maintenance_execution::reconcile_previous(state, claim).await?;
+    let started = std::time::Instant::now();
+    let outcome = crate::maintenance_execution::run(state, claim, async {
+        let mut store =
+            RolloutStore::open_existing_with_options(&uri, state.rollout_store_options())
+                .await
+                .map_err(|e| e.to_string())?;
+        let metrics = store
+            .commit_prepared_compaction(prepared)
+            .await
+            .map_err(|e| e.to_string())?;
+        update_stats_after_compaction(state, name, &store).await;
+        finish_compaction(state, &claim.task, metrics).await
+    })
+    .await;
+    metrics::histogram!("master_compaction_phase_duration_seconds", "phase" => "commit")
+        .record(started.elapsed().as_secs_f64());
+    outcome
+}
+
 /// Compact one experiment. The task-store claim owns the per-experiment write
 /// lock for the full execution.
 ///
@@ -337,6 +485,14 @@ async fn run_compaction(state: &Arc<MasterState>, task: &TaskRecord) -> Result<S
         return Err(format!("compaction is not scheduled for {kind:?} stores"));
     }
     let metrics = compact_inner(state, name).await?;
+    finish_compaction(state, task, metrics).await
+}
+
+async fn finish_compaction(
+    state: &Arc<MasterState>,
+    task: &TaskRecord,
+    metrics: lance::dataset::optimize::CompactionMetrics,
+) -> Result<String, String> {
     if state.config.index_after_compaction && metrics.fragments_added > 0 {
         // Best-effort: the compaction itself is done; the next compaction of
         // this target re-enqueues the index anyway.
@@ -390,12 +546,6 @@ async fn compact_inner(
 ) -> Result<lance::dataset::optimize::CompactionMetrics, String> {
     let uri = state.rollout_uri(name);
     let config = state.compaction_config();
-    let _permit = state
-        .compaction_permits
-        .clone()
-        .acquire_owned()
-        .await
-        .map_err(|_| "compaction semaphore closed".to_string())?;
 
     let mut store = RolloutStore::open_existing_with_options(&uri, state.rollout_store_options())
         .await
@@ -862,16 +1012,17 @@ pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
     // FIFO order within a kind is unchanged.
     let general = spawn_pool_poller(
         state.clone(),
-        sem,
+        sem.clone(),
         if merge_sem.is_some() {
-            TaskKinds::GENERAL
+            TaskKinds::GENERAL.without_compact()
         } else {
             // No separate merge budget: the general pool runs everything, so it
             // must still be allowed to claim MergeWal.
-            TaskKinds::ANY
+            TaskKinds::ANY.without_compact()
         },
         true,
     );
+    spawn_pool_poller(state.clone(), sem, TaskKinds::COMPACT, false);
     if let Some(merge) = merge_sem {
         spawn_pool_poller(state.clone(), merge, TaskKinds::MERGE_WAL, false);
     }
@@ -897,30 +1048,33 @@ fn spawn_pool_poller(
                 }
             }
             while pool.available_permits() > 0 {
+                // Reserve both budgets before taking durable table ownership.
+                let compact_permit = if kinds == TaskKinds::COMPACT {
+                    match state.compaction_permits.clone().try_acquire_owned() {
+                        Ok(permit) => Some(permit),
+                        Err(_) => break,
+                    }
+                } else {
+                    None
+                };
+                let Ok(permit) = pool.clone().try_acquire_owned() else {
+                    break;
+                };
                 let claim_start = std::time::Instant::now();
                 match state.task_store.claim_next_of_kinds(kinds).await {
                     Ok(Some(claim)) => {
                         let claim_elapsed = claim_start.elapsed();
-                        // The task is already claimed at this point (queue key
-                        // deleted, lease granted, target lock held), so time
-                        // spent here is a claimed-but-idle task holding its
-                        // per-experiment lock — worth seeing separately. This
-                        // loop only claims kinds its own pool runs, so the wait
-                        // is now bounded by that pool's own occupancy.
-                        let permit_start = std::time::Instant::now();
-                        let permit = pool
-                            .clone()
-                            .acquire_owned()
-                            .await
-                            .expect("semaphore never closed");
+                        // Local capacity was reserved before the claim. No
+                        // claimed table waits on a compaction semaphore.
                         let timing = TaskClaimTiming {
                             claim: claim_elapsed,
-                            permit_wait: permit_start.elapsed(),
+                            permit_wait: Duration::ZERO,
                         };
                         let st = state.clone();
                         tokio::spawn(async move {
                             run_task(&st, claim, timing).await;
                             drop(permit);
+                            drop(compact_permit);
                         });
                     }
                     Ok(None) => break,
@@ -1000,6 +1154,215 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn prepared_compact_yields_capacity_without_cancelling_busy_writer() {
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.maintenance.compaction_prepare_targets = vec!["exp".into()];
+        cfg.maintenance.compaction_commit_wait_secs = 0;
+        let state = MasterState::new(cfg).await.unwrap();
+        enqueue(&state, TaskKind::Compact, "exp").await.unwrap();
+        let mut prep = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap();
+        let merge = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let merge_id = merge.task.id.clone();
+        let error = wait_for_compaction_commit(&state, &mut prep)
+            .await
+            .unwrap_err();
+        assert!(error.contains("wait budget exhausted"));
+        state.task_store.finish(prep, Err(error)).await.unwrap();
+        assert_eq!(
+            state
+                .task_store
+                .get(&merge_id)
+                .await
+                .unwrap()
+                .unwrap()
+                .state,
+            TaskState::Running
+        );
+        // The other writer still owns its exact claim and can finish normally.
+        state
+            .task_store
+            .finish(merge, Ok("writer kept progressing".into()))
+            .await
+            .unwrap();
+        assert_eq!(
+            state
+                .task_store
+                .get(&merge_id)
+                .await
+                .unwrap()
+                .unwrap()
+                .state,
+            TaskState::Done
+        );
+        enqueue(&state, TaskKind::Compact, "exp").await.unwrap();
+        let mut next = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        wait_for_compaction_commit(&state, &mut next).await.unwrap();
+        state
+            .task_store
+            .finish(next, Ok("retry acquired free writer".into()))
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn compact_capacity_is_reserved_before_claim_without_blocking_other_kinds() {
+        let dir = TempDir::new().unwrap();
+        let state = MasterState::new(config(&dir)).await.unwrap();
+        RolloutStore::open(&state.rollout_uri("exp")).await.unwrap();
+        let occupied = state
+            .compaction_permits
+            .clone()
+            .acquire_owned()
+            .await
+            .unwrap();
+        let compact = enqueue(&state, TaskKind::Compact, "exp").await.unwrap();
+        enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap();
+        enqueue(&state, TaskKind::IndexId, "other").await.unwrap();
+        let pool = Arc::new(Semaphore::new(2));
+        let poller = spawn_pool_poller(state.clone(), pool.clone(), TaskKinds::COMPACT, false);
+        tokio::time::sleep(Duration::from_millis(650)).await;
+        assert_eq!(
+            state
+                .task_store
+                .get(&compact.id)
+                .await
+                .unwrap()
+                .unwrap()
+                .state,
+            TaskState::Queued
+        );
+        assert_eq!(pool.available_permits(), 2);
+        let merge = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let index = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::GENERAL.without_compact())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(index.task.kind, TaskKind::IndexId);
+        state
+            .task_store
+            .finish(merge, Ok("merge is not pinned by idle compact".into()))
+            .await
+            .unwrap();
+        state
+            .task_store
+            .finish(index, Ok("index remains dispatchable".into()))
+            .await
+            .unwrap();
+        drop(occupied);
+        tokio::time::timeout(Duration::from_secs(15), async {
+            loop {
+                let task = state.task_store.get(&compact.id).await.unwrap().unwrap();
+                if task.state == TaskState::Done {
+                    break;
+                }
+                assert_ne!(task.state, TaskState::Failed, "{:?}", task.error);
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        })
+        .await
+        .unwrap();
+        poller.abort();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn prepared_compact_runs_through_owned_commit_and_releases_both_claims() {
+        prepared_compact_commit_and_release(true).await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn prepared_compact_runs_through_legacy_commit_and_releases_both_claims() {
+        prepared_compact_commit_and_release(false).await;
+    }
+
+    async fn prepared_compact_commit_and_release(owned: bool) {
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.maintenance.compaction_prepare_targets = vec!["exp".into()];
+        if !owned {
+            cfg.merge_rollout.owned_targets.clear();
+        }
+        let state = MasterState::new(cfg).await.unwrap();
+        let uri = state.rollout_uri("exp");
+        let mut store = RolloutStore::open(&uri).await.unwrap();
+        for id in ["a", "b", "c", "d"] {
+            store.add(&[rollout_record(id)]).await.unwrap();
+            store.cleanup_own_shard().await.unwrap();
+        }
+        let task = enqueue(&state, TaskKind::Compact, "exp").await.unwrap();
+        let claim = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(claim.preparing_compaction());
+        run_task(
+            &state,
+            claim,
+            TaskClaimTiming {
+                claim: Duration::ZERO,
+                permit_wait: Duration::ZERO,
+            },
+        )
+        .await;
+        let finished = state.task_store.get(&task.id).await.unwrap().unwrap();
+        assert_eq!(finished.state, TaskState::Done, "{:?}", finished.error);
+        assert!(state
+            .task_store
+            .merge_coordinator()
+            .get("exp")
+            .await
+            .unwrap()
+            .is_none());
+        enqueue(&state, TaskKind::MergeWal, "exp").await.unwrap();
+        let merge = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        state
+            .task_store
+            .finish(merge, Ok("resumed".into()))
+            .await
+            .unwrap();
+        let reader = RolloutStore::open_existing_with_options(&uri, Default::default())
+            .await
+            .unwrap();
+        for id in ["a", "b", "c", "d"] {
+            assert!(reader.get_by_id(id).await.unwrap().is_some());
+        }
+    }
+
     #[tokio::test]
     #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn compact_noop_survives_restart_and_changes_invalidate_it() {
diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs
index ddedd35..fe63b77 100644
--- a/crates/lance-context-master/src/task_store.rs
+++ b/crates/lance-context-master/src/task_store.rs
@@ -54,6 +54,17 @@ impl TaskKinds {
         index_id: true,
     };
 
+    pub const COMPACT: Self = Self {
+        compact: true,
+        merge_wal: false,
+        index_id: false,
+    };
+
+    pub fn without_compact(mut self) -> Self {
+        self.compact = false;
+        self
+    }
+
     #[must_use]
     pub fn contains(self, kind: TaskKind) -> bool {
         match kind {
@@ -107,6 +118,8 @@ struct EtcdTaskStore {
     client: Client,
     prefix: String,
     lease_ttl: i64,
+    prepare_targets: Vec<String>,
+    rollout: lance_context_merge::rollout::MergeRollout,
 }
 
 /// Ownership of one running task. Dropping the claim stops lease renewal; etcd
@@ -116,11 +129,18 @@ pub struct TaskClaim {
     backend: ClaimBackend,
 }
 
+impl TaskClaim {
+    pub(crate) fn preparing_compaction(&self) -> bool {
+        self.backend.preparation_key.is_some() && self.backend.target_key.is_none()
+    }
+}
+
 struct ClaimBackend {
     token: String,
     lease_id: i64,
     claim_key: String,
     target_key: Option<String>,
+    preparation_key: Option<String>,
     keepalive: LeaseKeepalive,
 }
 
@@ -148,6 +168,111 @@ impl Drop for LeaseKeepalive {
 }
 
 impl TaskStore {
+    pub(crate) async fn preparation_owned(&self, claim: &TaskClaim) -> lance::Result<bool> {
+        let Some(key) = &claim.backend.preparation_key else {
+            return Ok(false);
+        };
+        let values = self
+            .inner
+            .read_values(&[claim.backend.claim_key.clone(), key.clone()])
+            .await?;
+        Ok(values
+            .iter()
+            .all(|v| v.as_deref() == Some(claim.backend.token.as_bytes())))
+    }
+
+    /// Acquire the existing table writer protocol only after immutable files
+    /// are ready. An ambiguous response can adopt only this claim's own token.
+    pub(crate) async fn promote_compaction(&self, claim: &mut TaskClaim) -> lance::Result<bool> {
+        assert!(claim.preparing_compaction());
+        let b = &claim.backend;
+        let target = &claim.task.target;
+        let execution_key = lance_context_merge::execution_key(&self.inner.prefix, target);
+        let target_key = self.inner.target_lock_key(target);
+        let merge_key = execution_key.replace("/merge-executions/", "/merge-claims/");
+        let active_key = crate::catchup::store::active_key(&self.inner.prefix, target);
+        let values = self
+            .inner
+            .read_values(&[
+                execution_key.clone(),
+                target_key.clone(),
+                merge_key.clone(),
+                active_key.clone(),
+            ])
+            .await?;
+        let old: Option<lance_context_merge::Execution> = values[0]
+            .as_deref()
+            .map(serde_json::from_slice)
+            .transpose()
+            .map_err(|e| lance::Error::io(e.to_string()))?;
+        if values[2].is_some()
+            || values[3].is_some()
+            || old.as_ref().is_some_and(|e| e.maintenance.is_none())
+        {
+            return Ok(false);
+        }
+        let expected_owner = old.as_ref().map(lance_context_merge::execution_owner);
+        if values[1].as_deref() != expected_owner.as_deref().map(str::as_bytes) {
+            return Ok(false);
+        }
+        let mut compares = vec![
+            Compare::value(b.claim_key.as_str(), CompareOp::Equal, b.token.as_bytes()),
+            Compare::value(
+                b.preparation_key.as_ref().unwrap().as_str(),
+                CompareOp::Equal,
+                b.token.as_bytes(),
+            ),
+            Compare::version(merge_key.as_str(), CompareOp::Equal, 0),
+            Compare::version(active_key, CompareOp::Equal, 0),
+        ];
+        for (key, value) in [
+            (execution_key, &values[0]),
+            (target_key.clone(), &values[1]),
+        ] {
+            compares.push(match value {
+                Some(v) => Compare::value(key, CompareOp::Equal, v.clone()),
+                None => Compare::version(key, CompareOp::Equal, 0),
+            });
+        }
+        let opts = Some(PutOptions::new().with_lease(b.lease_id));
+        let mut writes = vec![TxnOp::put(
+            merge_key.clone(),
+            b.token.as_bytes(),
+            opts.clone(),
+        )];
+        if old.is_none() {
+            writes.push(TxnOp::put(target_key.clone(), b.token.as_bytes(), opts));
+        }
+        let response = self
+            .inner
+            .client
+            .clone()
+            .txn(Txn::new().when(compares).and_then(writes))
+            .await;
+        let accepted = match response {
+            Ok(r) => r.succeeded(),
+            Err(error) => {
+                let values = self
+                    .inner
+                    .read_values(&[b.claim_key.clone(), merge_key, target_key.clone()])
+                    .await?;
+                if values[0].as_deref() == Some(b.token.as_bytes())
+                    && values[1].as_deref() == Some(b.token.as_bytes())
+                    && values[2].as_deref()
+                        == Some(expected_owner.as_deref().unwrap_or(&b.token).as_bytes())
+                {
+                    true
+                } else {
+                    return Err(etcd_error("promote compaction")(error));
+                }
+            }
+        };
+        if accepted {
+            claim.backend.target_key = Some(target_key);
+        }
+        Ok(accepted)
+    }
+
     pub(crate) fn merge_coordinator(&self) -> lance_context_merge::Coordinator {
         lance_context_merge::Coordinator::new(self.inner.client.clone(), self.inner.prefix.clone())
     }
@@ -525,6 +650,8 @@ impl EtcdTaskStore {
             client,
             prefix: config.etcd.prefix().to_string(),
             lease_ttl: config.etcd_lease_ttl_secs,
+            prepare_targets: config.maintenance.compaction_prepare_targets.clone(),
+            rollout: config.merge_rollout.clone(),
         })
     }
 
@@ -677,6 +804,21 @@ impl EtcdTaskStore {
                     }
                 }
 
+                let preparing = task.kind == TaskKind::Compact
+                    && !task.target.starts_with("generic:")
+                    && !self.rollout.draining(&task.target)
+                    && self
+                        .prepare_targets
+                        .iter()
+                        .any(|t| t == "*" || t == &task.target);
+                let write_claim = requires_target_lock(task.kind) && !preparing;
+                let preparation_key = preparing.then(|| {
+                    format!(
+                        "{}/compact-preparations/{}",
+                        self.prefix,
+                        encode_segment(&task.target)
+                    )
+                });
                 let execution_key = lance_context_merge::execution_key(&self.prefix, &task.target);
                 let snapshot = self
                     .read_values(&[
@@ -685,6 +827,13 @@ impl EtcdTaskStore {
                         crate::catchup::store::active_key(&self.prefix, &task.target),
                         execution_key.replace("/merge-executions/", "/merge-claims/"),
                         self.claim_key(&task.id),
+                        preparation_key.clone().unwrap_or_else(|| {
+                            format!(
+                                "{}/compact-preparations/{}",
+                                self.prefix,
+                                encode_segment(&task.target)
+                            )
+                        }),
                     ])
                     .await?;
                 let merge_execution: Option<lance_context_merge::Execution> = snapshot[0]
@@ -699,7 +848,8 @@ impl EtcdTaskStore {
                     .map(lance_context_merge::execution_owner);
                 let expected_job = only_id.map(|(_, job)| job.as_bytes());
                 if snapshot[4].is_some()
-                    || (requires_target_lock(task.kind)
+                    || (preparing && (snapshot[5].is_some() || snapshot[2].is_some()))
+                    || (write_claim
                         && (snapshot[2].as_deref() != expected_job
                             || snapshot[3].is_some()
                             || snapshot[1].as_deref()
@@ -711,7 +861,8 @@ impl EtcdTaskStore {
                 // Only MergeWal reconciles a worker execution. A local fenced
                 // mutation may be recovered by any table writer after the
                 // exclusive reconciler lease has expired.
-                if task.kind != TaskKind::MergeWal
+                if !preparing
+                    && task.kind != TaskKind::MergeWal
                     && merge_execution
                         .as_ref()
                         .is_some_and(|e| e.maintenance.is_none())
@@ -730,8 +881,7 @@ impl EtcdTaskStore {
                     }
                 };
                 let claim_key = self.claim_key(&task.id);
-                let target_key =
-                    requires_target_lock(task.kind).then(|| self.target_lock_key(&task.target));
+                let target_key = write_claim.then(|| self.target_lock_key(&task.target));
                 let queue_key = self.queue_key(&task.id);
                 let running_key = self.running_key(&task.id);
                 let queued_value = encode_task(&task)?;
@@ -743,8 +893,16 @@ impl EtcdTaskStore {
                     Compare::version(claim_key.as_str(), CompareOp::Equal, 0),
                     Compare::version(queue_key.as_str(), CompareOp::Greater, 0),
                 ];
+                if let Some(key) = &preparation_key {
+                    compares.push(Compare::version(key.as_str(), CompareOp::Equal, 0));
+                    compares.push(Compare::version(
+                        crate::catchup::store::active_key(&self.prefix, &task.target),
+                        CompareOp::Equal,
+                        0,
+                    ));
+                }
                 let catchup_key = crate::catchup::store::active_key(&self.prefix, &task.target);
-                if requires_target_lock(task.kind) {
+                if write_claim {
                     compares.push(match only_id {
                         Some((_, job)) => Compare::value(catchup_key, CompareOp::Equal, job),
                         None => Compare::version(catchup_key, CompareOp::Equal, 0),
@@ -767,7 +925,7 @@ impl EtcdTaskStore {
                 let merge_claim_key =
                     lance_context_merge::execution_key(&self.prefix, &task.target)
                         .replace("/merge-executions/", "/merge-claims/");
-                if requires_target_lock(task.kind) {
+                if write_claim {
                     compares.push(Compare::version(
                         merge_claim_key.as_str(),
                         CompareOp::Equal,
@@ -781,7 +939,14 @@ impl EtcdTaskStore {
                     TxnOp::put(running_key, running_value, None),
                     TxnOp::put(claim_key.as_str(), token.as_bytes(), lease_options.clone()),
                 ];
-                if requires_target_lock(task.kind) {
+                if let Some(key) = &preparation_key {
+                    operations.push(TxnOp::put(
+                        key.as_str(),
+                        token.as_bytes(),
+                        lease_options.clone(),
+                    ));
+                }
+                if write_claim {
                     operations.push(TxnOp::put(
                         merge_claim_key,
                         token.as_bytes(),
@@ -821,6 +986,7 @@ impl EtcdTaskStore {
                                 lease_id,
                                 claim_key,
                                 target_key,
+                                preparation_key,
                                 keepalive,
                             },
                         }),
@@ -847,10 +1013,12 @@ impl EtcdTaskStore {
             lease_id,
             claim_key,
             target_key,
+            preparation_key,
             keepalive,
         } = claim.backend;
         let mut task = claim.task;
-        let unresolved = if requires_target_lock(task.kind) {
+        let owns_table = target_key.is_some();
+        let unresolved = if owns_table {
             lance_context_merge::Coordinator::new(self.client.clone(), self.prefix.clone())
                 .get(&task.target)
                 .await
@@ -864,13 +1032,16 @@ impl EtcdTaskStore {
             TxnOp::delete(claim_key.as_str(), None),
             TxnOp::delete(self.running_key(&task.id), None),
         ];
-        if requires_target_lock(task.kind) {
+        if owns_table {
             operations.push(TxnOp::delete(
                 lance_context_merge::execution_key(&self.prefix, &task.target)
                     .replace("/merge-executions/", "/merge-claims/"),
                 None,
             ));
         }
+        if let Some(key) = preparation_key {
+            operations.push(TxnOp::delete(key, None));
+        }
         if unresolved.is_none() {
             if let Some(key) = &target_key {
                 operations.push(TxnOp::delete(key.as_str(), None));
@@ -879,28 +1050,28 @@ impl EtcdTaskStore {
         if let Some(key) = self.dedupe_key(task.kind, &task.target, &task.depends_on) {
             operations.push(TxnOp::delete(key, None));
         }
+        let mut compares = vec![Compare::value(
+            claim_key.as_str(),
+            CompareOp::Equal,
+            token.as_bytes(),
+        )];
+        if owns_table {
+            compares.push(match &unresolved {
+                Some(execution) => Compare::value(
+                    lance_context_merge::execution_key(&self.prefix, &task.target),
+                    CompareOp::Equal,
+                    serde_json::to_vec(execution).map_err(|e| lance::Error::io(e.to_string()))?,
+                ),
+                None => Compare::version(
+                    lance_context_merge::execution_key(&self.prefix, &task.target),
+                    CompareOp::Equal,
+                    0,
+                ),
+            });
+        }
         let mut client = self.client.clone();
         let completed = client
-            .txn(
-                Txn::new()
-                    .when([
-                        Compare::value(claim_key.as_str(), CompareOp::Equal, token.as_bytes()),
-                        match &unresolved {
-                            Some(execution) => Compare::value(
-                                lance_context_merge::execution_key(&self.prefix, &task.target),
-                                CompareOp::Equal,
-                                serde_json::to_vec(execution)
-                                    .map_err(|e| lance::Error::io(e.to_string()))?,
-                            ),
-                            None => Compare::version(
-                                lance_context_merge::execution_key(&self.prefix, &task.target),
-                                CompareOp::Equal,
-                                0,
-                            ),
-                        },
-                    ])
-                    .and_then(operations),
-            )
+            .txn(Txn::new().when(compares).and_then(operations))
             .await
             .map_err(etcd_error("complete task"))?
             .succeeded();
@@ -2098,6 +2269,209 @@ mod tests {
             .unwrap();
     }
 
+    async fn preparation_test_store() -> (TempDir, MasterConfig, TaskStore) {
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS")
+            .unwrap()
+            .split(',')
+            .map(str::to_string)
+            .collect();
+        cfg.etcd.etcd_prefix = format!("/lance-context/test/{}", generate_id());
+        cfg.maintenance.compaction_prepare_targets = vec!["*".into()];
+        let store = TaskStore::open(&cfg).await.unwrap();
+        (dir, cfg, store)
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn compaction_preparation_allows_merge_but_commit_is_exclusive() {
+        let (_dir, _cfg, store) = preparation_test_store().await;
+        store
+            .enqueue(TaskKind::Compact, "hot", Vec::new())
+            .await
+            .unwrap();
+        let mut prep = store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(prep.preparing_compaction());
+        assert!(store
+            .inner
+            .get_text(&store.inner.target_lock_key("hot"))
+            .await
+            .unwrap()
+            .is_none());
+        store
+            .enqueue(TaskKind::MergeWal, "hot", Vec::new())
+            .await
+            .unwrap();
+        let merge = store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(!store.promote_compaction(&mut prep).await.unwrap());
+        store
+            .finish(merge, Ok("append while rewriting".into()))
+            .await
+            .unwrap();
+        assert!(store.promote_compaction(&mut prep).await.unwrap());
+        assert!(!prep.preparing_compaction());
+        store
+            .enqueue(TaskKind::MergeWal, "hot", Vec::new())
+            .await
+            .unwrap();
+        assert!(store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .is_none());
+        store
+            .finish(prep, Ok("compact published".into()))
+            .await
+            .unwrap();
+        let next = store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        store.finish(next, Ok("continued".into())).await.unwrap();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn preparation_and_promotion_respect_dedicated_owner() {
+        let (_dir, cfg, store) = preparation_test_store().await;
+        store
+            .enqueue(TaskKind::Compact, "hot", Vec::new())
+            .await
+            .unwrap();
+        let active = crate::catchup::store::active_key(&cfg.etcd.etcd_prefix, "hot");
+        store
+            .inner
+            .client
+            .clone()
+            .put(active.clone(), "dedicated", None)
+            .await
+            .unwrap();
+        assert!(store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .is_none());
+        store
+            .inner
+            .client
+            .clone()
+            .delete(active.clone(), None)
+            .await
+            .unwrap();
+        let mut prep = store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        store
+            .inner
+            .client
+            .clone()
+            .put(active.clone(), "new-dedicated", None)
+            .await
+            .unwrap();
+        assert!(!store.promote_compaction(&mut prep).await.unwrap());
+        store.finish(prep, Err("superseded".into())).await.unwrap();
+        assert_eq!(
+            store.inner.get_text(&active).await.unwrap().as_deref(),
+            Some("new-dedicated")
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn failed_preparation_cannot_release_an_active_merge() {
+        let (_dir, _cfg, store) = preparation_test_store().await;
+        store
+            .enqueue(TaskKind::Compact, "hot", Vec::new())
+            .await
+            .unwrap();
+        let prep = store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        store
+            .enqueue(TaskKind::MergeWal, "hot", Vec::new())
+            .await
+            .unwrap();
+        let merge = store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let coordinator = store.merge_coordinator();
+        let execution = lance_context_merge::Execution::new("hot", "worker", &merge.task.id, 600);
+        assert!(coordinator
+            .reserve(&store.merge_claim(&merge), &execution)
+            .await
+            .unwrap());
+        let running = coordinator.start(&execution).await.unwrap().unwrap();
+        store
+            .finish(prep, Err("preparation failed".into()))
+            .await
+            .unwrap();
+        assert_eq!(coordinator.get("hot").await.unwrap(), Some(running.clone()));
+        assert_eq!(
+            store
+                .inner
+                .get_text(&store.inner.target_lock_key("hot"))
+                .await
+                .unwrap(),
+            Some(lance_context_merge::execution_owner(&running))
+        );
+        assert!(coordinator.finish(&running, Ok(1)).await.unwrap());
+        let terminal = coordinator.get("hot").await.unwrap().unwrap();
+        assert!(coordinator
+            .release(&store.merge_claim(&merge), &terminal)
+            .await
+            .unwrap());
+        store.finish(merge, Ok("done".into())).await.unwrap();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn expired_preparation_cannot_promote_or_finish_replacement() {
+        let (_dir, _cfg, store) = preparation_test_store().await;
+        let task = store
+            .enqueue(TaskKind::Compact, "hot", Vec::new())
+            .await
+            .unwrap();
+        let mut old = store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        store
+            .inner
+            .revoke_lease(old.backend.lease_id)
+            .await
+            .unwrap();
+        store.recover_orphaned().await.unwrap();
+        let mut replacement = store
+            .claim_next_of_kinds(TaskKinds::COMPACT)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(replacement.task.id, task.id);
+        assert!(!store.preparation_owned(&old).await.unwrap());
+        assert!(!store.promote_compaction(&mut old).await.unwrap());
+        assert!(store.finish(old, Ok("stale".into())).await.is_err());
+        assert!(store.preparation_owned(&replacement).await.unwrap());
+        assert!(store.promote_compaction(&mut replacement).await.unwrap());
+        store.finish(replacement, Ok("valid".into())).await.unwrap();
+    }
+
     #[tokio::test]
     async fn connect_requires_etcd_endpoints() {
         let dir = TempDir::new().unwrap();
diff --git a/docs/concurrent-compaction.md b/docs/concurrent-compaction.md
new file mode 100644
index 0000000..f9ff1b1
--- /dev/null
+++ b/docs/concurrent-compaction.md
@@ -0,0 +1,57 @@
+# Compaction preparation alongside WAL merge
+
+A large compaction can spend many minutes reading and encoding old fragments.
+Holding the table writer lock during that work prevents WAL publication even
+when the merge only appends new fragments. `COMPACTION_PREPARE_TARGETS` selects
+rollout targets (comma-separated names, or `*`) for two-phase compaction. The
+default empty list preserves the previous path; generic stores and draining
+targets do not use preparation.
+
+1. Claim the durable Compact task and a leased `compact-preparations/<target>`
+   key. Do not acquire the shared table lock, merge claim, or execution record.
+   The normal task lease/recovery protocol remains unchanged.
+2. Read a fixed snapshot and write immutable, unpublished rewrite files. A
+   pinned commit authorizer rejects any manifest publication during this phase.
+   Actual storage/processing progress drives the idle watchdog; claim loss
+   cancels preparation. Neither cancellation nor process loss changes live data.
+3. Once files are ready, acquire the normal table lock and merge-claim key using
+   one transaction that verifies the original task/preparation tokens and all
+   current execution, table, and dedicated-owner predicates. A lost response
+   may only be adopted after verifying this task's unique token.
+4. Reconcile prior maintenance, then open a fresh handle in the existing
+   maintenance execution scope. Owned targets retain version fencing and
+   ambiguous-commit recovery. Unowned targets retain their existing commit
+   protocol. Validate source fragments and schema, then call Lance's
+   `commit_compaction` with the original RewriteResult read versions. Fragment
+   reservations, index remapping and final publication all occur in this phase.
+5. Finish the task and release its own keys. A failed preparation must never
+   delete another task's merge claim or execution record.
+
+An append-only merge does not alter the selected source fragments and can
+complete during preparation. An update, delete, overlapping rewrite, or schema
+change may invalidate the prepared output; stale files must not resurrect old
+rows. Lance retains its intervening-transaction conflict checks in addition to
+the source-fragment preflight. WAL merged-generation watermarks remain part of
+the latest manifest and must survive either commit order.
+
+Compaction capacity is reserved before claiming, together with the existing
+general task budget. Index/repair and WAL polling remain independent. The
+`master_compaction_phase_duration_seconds` histogram separates `prepare`,
+`commit_wait`, and `commit`; task/target identities appear only in logs.
+
+Prepared output is deliberately not a new durable publication protocol. A
+process lost before commit may leave unreferenced immutable files and its task
+is requeued after lease expiry. Files are never considered committed because
+encoding completed. A newly selected dedicated owner still excludes compaction
+publication. Preparation does not override maintenance dependencies or guards.
+`COMPACTION_COMMIT_WAIT_SECS` (default 120) bounds how long ready output retains
+local capacity while waiting for a writer. Exhaustion fails only this compact
+attempt, releases its preparation claim, and leaves the progressing writer
+untouched. A later compact replans; no prepared output is treated as committed.
+This admission budget is separate from the progress-based execution watchdog.
+
+Deployment can mix old and new masters: queue/task records retain their schema,
+and an expired preparation claim can be safely retried through either path.
+Disabling the option affects newly claimed tasks; a live preparation can finish
+through the same fenced publication path. Do not manually delete ownership or
+restart a progressing compaction to toggle this option.