diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 427e98104..f50293ef0 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -57,6 +57,9 @@ class Split:
     def serialize(self) -> bytes: ...
 
 class Plan:
+    def __init__(self, snapshot_id: Optional[int], splits: Iterable[Split]) -> None:
+        """Capture typed split metadata; constructing a plan does not authorize execution."""
+        ...
     def snapshot_id(self) -> Optional[int]:
         """Selected snapshot, preserved for empty plans; None if no snapshot was selected."""
         ...
@@ -66,6 +69,16 @@ class Plan:
         ...
     def __len__(self) -> int: ...
 
+class RowIdPlan:
+    def __init__(self, plan: Plan) -> None:
+        """Group complete data-evolution files from plan_all without reading manifests."""
+        ...
+    def snapshot_id(self) -> Optional[int]: ...
+    def groups(self) -> List[Tuple[int, int, Split]]:
+        """Inclusive row-ID bounds and all physical files of each logical group."""
+        ...
+    def plan(self) -> Plan: ...
+
 class ScanStats:
     """Read-only counters from normal metadata planning."""
     @property
@@ -417,6 +430,14 @@ class Table:
         """
         ...
 
+class GlobalIndexBuildPlan:
+    def snapshot_id(self) -> Optional[int]: ...
+    def shards(self) -> List["GlobalIndexBuildShard"]: ...
+
+class GlobalIndexBuildShard:
+    def snapshot_id(self) -> int: ...
+    def row_range(self) -> Tuple[int, int]: ...
+
 class GlobalIndexBuildBuilder:
     @staticmethod
     def supports_index_type(index_type: str) -> bool: ...
@@ -425,6 +446,16 @@ class GlobalIndexBuildBuilder:
     def with_index_type(self, index_type: str) -> "GlobalIndexBuildBuilder": ...
     def with_options(self, options: Dict[str, str]) -> "GlobalIndexBuildBuilder": ...
     def with_partition_filter(self, predicate: Dict[str, Any]) -> "GlobalIndexBuildBuilder": ...
+    def plan(self) -> GlobalIndexBuildPlan:
+        """Capture independently executable vindex shards at one snapshot."""
+        ...
+    def build_shard(self, shard: GlobalIndexBuildShard) -> List["CommitMessage"]:
+        """Build one captured shard with fresh worker authorization.
+
+        The shard supplies its schema, index configuration and options;
+        this builder supplies the worker table and I/O environment.
+        """
+        ...
     def build(self) -> List["CommitMessage"]:
         """Build unpublished index files; the caller decides whether to commit or abort."""
         ...
@@ -477,6 +508,17 @@ class BatchTableWrite:
         ...
 
 class BatchTableUpdate:
+    def with_plan(self, plan: Plan) -> "BatchTableUpdate":
+        """Reuse complete plan_all file groups for row-ID updates and deletes.
+
+        Workers reauthorize against the table; a plan is metadata, not a grant.
+        Data manifests are not replanned. Deletes still read the captured index
+        manifest to preserve deletion vectors for every file in the bucket.
+        Assign all deletes for one partition/bucket to the same worker.
+        Selected row ranges, scores and streaming splits are not mutation plans.
+        Other update operations reject this configuration before consuming input.
+        """
+        ...
     def with_blob_uri_reader_factory(
         self, factory: Optional[UriReaderFactory]
     ) -> "BatchTableUpdate": ...
@@ -540,6 +582,17 @@ class TableUpdateByRowId:
     def _abort(self) -> None: ...
 
 class StreamTableUpdate:
+    def with_plan(self, plan: Plan) -> "StreamTableUpdate":
+        """Reuse complete plan_all file groups for row-ID updates and deletes.
+
+        Workers reauthorize against the table; a plan is metadata, not a grant.
+        Data manifests are not replanned. Deletes still read the captured index
+        manifest to preserve deletion vectors for every file in the bucket.
+        Assign all deletes for one partition/bucket to the same worker.
+        Selected row ranges, scores and streaming splits are not mutation plans.
+        Other update operations reject this configuration before consuming input.
+        """
+        ...
     def with_blob_uri_reader_factory(
         self, factory: Optional[UriReaderFactory]
     ) -> "StreamTableUpdate": ...
@@ -707,6 +760,11 @@ class BatchTableCommit:
     ) -> None:
         """Commit once, using the builder's overwrite configuration and batch identifier."""
         ...
+    def check_row_id_conflicts(
+        self, messages: Iterable[CommitMessage], latest_snapshot_id: int, check_from_snapshot: int
+    ) -> None:
+        """Check captured DML history without publishing; compactions are ignored."""
+        ...
     def truncate_table(self) -> None:
         """Truncate all data; shares the one-time guard with commit()."""
         ...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 7e56cedaf..33fdd2659 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -993,6 +993,8 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     this.add_class::<crate::table::PyTable>()?;
     this.add_class::<crate::system_table::PySystemRead>()?;
     this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
+    this.add_class::<crate::index_build::PyGlobalIndexBuildPlan>()?;
+    this.add_class::<crate::index_build::PyGlobalIndexBuildShard>()?;
     this.add_class::<crate::index_build::PyGlobalIndexDropBuilder>()?;
     this.add_class::<crate::full_text_search::PyFullTextSearchBuilder>()?;
     this.add_class::<crate::full_text_search::PyFullTextScan>()?;
@@ -1019,6 +1021,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     this.add_class::<crate::read::PyTableScan>()?;
     this.add_class::<crate::read::PyStreamTableScan>()?;
     this.add_class::<crate::read::PyPlan>()?;
+    this.add_class::<crate::read::PyRowIdPlan>()?;
     this.add_class::<crate::read::PyFileScanPlan>()?;
     this.add_class::<crate::read::PyScanStats>()?;
     this.add_class::<crate::read::PyTableRead>()?;
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index 4e9b277a5..d5570ccd2 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -32,9 +32,20 @@ pyo3::create_exception!(
     "A catalog failure with code, resource_type, resource_name and message attributes."
 );
 
+pyo3::create_exception!(
+    pypaimon_rust,
+    RowIdExistenceConflict,
+    PyRuntimeError,
+    "A staged file no longer matches the current row-ID layout."
+);
+
 pub fn register_module(py: Python<'_>, module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add("ForkSafetyError", py.get_type::<ForkSafetyError>())?;
-    module.add("CatalogError", py.get_type::<CatalogError>())
+    module.add("CatalogError", py.get_type::<CatalogError>())?;
+    module.add(
+        "RowIdExistenceConflict",
+        py.get_type::<RowIdExistenceConflict>(),
+    )
 }
 
 /// Preserve the catalog error contract across the FFI without parsing display strings.
@@ -134,6 +145,28 @@ pub fn to_py_err(err: paimon::Error) -> PyErr {
     if err.is_process_fork_unsupported() {
         return ForkSafetyError::new_err(err.to_string());
     }
+    if let paimon::Error::RowIdExistenceConflict {
+        file_name,
+        first_row_id,
+        row_count,
+        bucket,
+        current_first_row_id,
+        current_row_count,
+    } = &err
+    {
+        return Python::attach(|py| {
+            let exception = RowIdExistenceConflict::new_err(err.to_string());
+            let value = exception.value(py);
+            value.setattr("file_name", file_name)?;
+            value.setattr("first_row_id", first_row_id)?;
+            value.setattr("row_count", row_count)?;
+            value.setattr("bucket", bucket)?;
+            value.setattr("current_first_row_id", current_first_row_id)?;
+            value.setattr("current_row_count", current_row_count)?;
+            Ok::<_, PyErr>(exception)
+        })
+        .unwrap_or_else(|error| error);
+    }
     match err {
         // Unimplemented scan semantics: distinct from malformed input so upper
         // layers can catch NotImplementedError and decide on a fallback.
@@ -170,6 +203,124 @@ pub fn df_to_py_err(err: datafusion::error::DataFusionError) -> PyErr {
 mod tests {
     use super::*;
 
+    #[test]
+    fn row_id_existence_error_preserves_structured_layout() {
+        Python::attach(|py| {
+            let error = to_py_err(paimon::Error::RowIdExistenceConflict {
+                file_name: "opaque '(4): file.parquet".into(),
+                first_row_id: Some(4),
+                row_count: 8,
+                bucket: 3,
+                current_first_row_id: None,
+                current_row_count: Some(9),
+            });
+            assert!(error.is_instance_of::<RowIdExistenceConflict>(py));
+            let value = error.value(py);
+            assert_eq!(
+                value
+                    .getattr("file_name")
+                    .unwrap()
+                    .extract::<String>()
+                    .unwrap(),
+                "opaque '(4): file.parquet"
+            );
+            assert_eq!(
+                value
+                    .getattr("first_row_id")
+                    .unwrap()
+                    .extract::<i64>()
+                    .unwrap(),
+                4
+            );
+            assert_eq!(
+                value
+                    .getattr("row_count")
+                    .unwrap()
+                    .extract::<i64>()
+                    .unwrap(),
+                8
+            );
+            assert_eq!(
+                value.getattr("bucket").unwrap().extract::<i32>().unwrap(),
+                3
+            );
+            assert!(value.getattr("current_first_row_id").unwrap().is_none());
+            assert_eq!(
+                value
+                    .getattr("current_row_count")
+                    .unwrap()
+                    .extract::<i64>()
+                    .unwrap(),
+                9
+            );
+        });
+    }
+
+    #[test]
+    fn row_id_existence_error_survives_python_worker_serialization() {
+        Python::attach(|py| {
+            let modules = py
+                .import("sys")
+                .unwrap()
+                .getattr("modules")
+                .unwrap()
+                .cast_into::<pyo3::types::PyDict>()
+                .unwrap();
+            let previous = modules.get_item("pypaimon_rust").unwrap();
+            let module = PyModule::new(py, "pypaimon_rust").unwrap();
+            register_module(py, &module).unwrap();
+            modules.set_item("pypaimon_rust", &module).unwrap();
+            let result = (|| -> PyResult<()> {
+                let error = to_py_err(paimon::Error::RowIdExistenceConflict {
+                    file_name: "worker:staged.parquet".into(),
+                    first_row_id: Some(4),
+                    row_count: 8,
+                    bucket: 3,
+                    current_first_row_id: None,
+                    current_row_count: Some(9),
+                });
+                let pickle = py.import("pickle")?;
+                let bytes = pickle.call_method1("dumps", (error.value(py),))?;
+                let restored = pickle.call_method1("loads", (bytes,))?;
+                assert!(restored.is_instance_of::<RowIdExistenceConflict>());
+                assert_eq!(
+                    restored.getattr("file_name")?.extract::<String>()?,
+                    "worker:staged.parquet"
+                );
+                assert_eq!(restored.getattr("first_row_id")?.extract::<i64>()?, 4);
+                assert_eq!(restored.getattr("row_count")?.extract::<i64>()?, 8);
+                assert_eq!(restored.getattr("bucket")?.extract::<i32>()?, 3);
+                assert!(restored.getattr("current_first_row_id")?.is_none());
+                assert_eq!(restored.getattr("current_row_count")?.extract::<i64>()?, 9);
+                Ok(())
+            })();
+            match previous {
+                Some(previous) => modules.set_item("pypaimon_rust", previous).unwrap(),
+                None => modules.del_item("pypaimon_rust").unwrap(),
+            }
+            result.unwrap();
+        });
+    }
+
+    #[test]
+    fn unknown_commit_outcome_does_not_expose_a_rewritable_conflict() {
+        Python::attach(|py| {
+            let error = to_py_err(paimon::Error::UnexpectedError {
+                message: "Commit outcome may be unknown; retain prepared files".into(),
+                source: Some(Box::new(paimon::Error::RowIdExistenceConflict {
+                    file_name: "staged.parquet".into(),
+                    first_row_id: Some(0),
+                    row_count: 10,
+                    bucket: 0,
+                    current_first_row_id: None,
+                    current_row_count: None,
+                })),
+            });
+            assert!(error.is_instance_of::<PyValueError>(py));
+            assert!(!error.is_instance_of::<RowIdExistenceConflict>(py));
+        });
+    }
+
     #[test]
     fn process_fork_error_has_distinct_python_type() {
         Python::attach(|py| {
diff --git a/bindings/python/src/index_build.rs b/bindings/python/src/index_build.rs
index ff5079c60..b9e016a60 100644
--- a/bindings/python/src/index_build.rs
+++ b/bindings/python/src/index_build.rs
@@ -21,10 +21,14 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use paimon::spec::Predicate;
-use paimon::table::{GlobalIndexBuildBuilder, GlobalIndexDropBuilder, Table};
+use paimon::table::{
+    GlobalIndexBuildBuilder, GlobalIndexBuildPlan, GlobalIndexBuildShard, GlobalIndexDropBuilder,
+    Table,
+};
 use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
-use pyo3::types::PyDict;
+use pyo3::types::{PyBytes, PyDict};
 
 use crate::error::to_py_err;
 use crate::predicate::dict_to_table_predicate;
@@ -35,6 +39,84 @@ enum PyIndexColumns {
     Columns(Vec<String>),
 }
 
+#[pyclass(
+    name = "GlobalIndexBuildPlan",
+    module = "pypaimon_rust.datafusion",
+    frozen
+)]
+pub struct PyGlobalIndexBuildPlan {
+    inner: GlobalIndexBuildPlan,
+}
+
+#[pyclass(
+    name = "GlobalIndexBuildShard",
+    module = "pypaimon_rust.datafusion",
+    frozen
+)]
+pub struct PyGlobalIndexBuildShard {
+    inner: GlobalIndexBuildShard,
+}
+
+// Runtime handles stay in the worker's Table; pickle carries domain metadata.
+macro_rules! build_plan_pickle {
+    ($class:ident, { $($methods:item)* }) => {
+        #[pymethods]
+        impl $class {
+            $($methods)*
+
+            fn _serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
+                let bytes = serde_json::to_vec(&self.inner)
+                    .map_err(|error| PyValueError::new_err(error.to_string()))?;
+                Ok(PyBytes::new(py, &bytes))
+            }
+
+            #[staticmethod]
+            fn _deserialize(state: &Bound<'_, PyBytes>) -> PyResult<Self> {
+                serde_json::from_slice(state.as_bytes())
+                    .map(|inner| Self { inner })
+                    .map_err(|error| {
+                        PyValueError::new_err(format!(
+                            "Invalid global index build metadata: {error}"
+                        ))
+                    })
+            }
+
+            fn __reduce__<'py>(
+                slf: &Bound<'py, Self>,
+                py: Python<'py>,
+            ) -> PyResult<(Py<PyAny>, (Py<PyBytes>,))> {
+                let bytes = slf.borrow()._serialize(py)?.unbind();
+                Ok((slf.get_type().getattr("_deserialize")?.unbind(), (bytes,)))
+            }
+        }
+    };
+}
+
+build_plan_pickle!(PyGlobalIndexBuildPlan, {
+    fn snapshot_id(&self) -> Option<i64> {
+        self.inner.snapshot_id()
+    }
+
+    fn shards(&self) -> Vec<PyGlobalIndexBuildShard> {
+        self.inner
+            .shards()
+            .into_iter()
+            .map(|inner| PyGlobalIndexBuildShard { inner })
+            .collect()
+    }
+});
+
+build_plan_pickle!(PyGlobalIndexBuildShard, {
+    fn snapshot_id(&self) -> i64 {
+        self.inner.snapshot_id()
+    }
+
+    fn row_range(&self) -> (i64, i64) {
+        let range = self.inner.row_range();
+        (range.from(), range.to())
+    }
+});
+
 #[pyclass(name = "GlobalIndexBuildBuilder", module = "pypaimon_rust.datafusion")]
 pub struct PyGlobalIndexBuildBuilder {
     table: Arc<Table>,
@@ -225,6 +307,26 @@ impl PyGlobalIndexBuildBuilder {
         Ok(messages.into_iter().map(PyCommitMessage::new).collect())
     }
 
+    fn plan(&self, py: Python<'_>) -> PyResult<PyGlobalIndexBuildPlan> {
+        let builder = self.builder().map_err(to_py_err)?;
+        let inner = py
+            .detach(|| runtime().block_on(builder.plan()))
+            .map_err(to_py_err)?;
+        Ok(PyGlobalIndexBuildPlan { inner })
+    }
+
+    fn build_shard(
+        &self,
+        py: Python<'_>,
+        shard: &PyGlobalIndexBuildShard,
+    ) -> PyResult<Vec<PyCommitMessage>> {
+        let builder = self.table.new_global_index_build_builder();
+        let messages = py
+            .detach(|| runtime().block_on(builder.build_shard(&shard.inner)))
+            .map_err(to_py_err)?;
+        Ok(messages.into_iter().map(PyCommitMessage::new).collect())
+    }
+
     fn execute(&self, py: Python<'_>) -> PyResult<usize> {
         let builder = self.builder().map_err(to_py_err)?;
         py.detach(|| runtime().block_on(builder.execute()))
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 05996167d..3545ff390 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -1170,7 +1170,7 @@ impl PyRecordBatchReader {
 
 #[pyclass(name = "Plan", module = "pypaimon_rust.datafusion")]
 pub struct PyPlan {
-    inner: paimon::table::Plan,
+    pub(crate) inner: paimon::table::Plan,
 }
 
 #[pyclass(name = "FileScanPlan", module = "pypaimon_rust.datafusion", frozen)]
@@ -1197,6 +1197,20 @@ impl From<paimon::table::Plan> for PyPlan {
 
 #[pymethods]
 impl PyPlan {
+    /// Restore captured planning metadata using typed splits. Execution
+    /// authorizes the transported plan independently.
+    #[new]
+    #[pyo3(signature = (snapshot_id, splits))]
+    fn new(snapshot_id: Option<i64>, splits: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let splits = splits
+            .try_iter()?
+            .map(|split| Ok(split?.extract::<PyRef<'_, PySplit>>()?.inner.clone()))
+            .collect::<PyResult<Vec<_>>>()?;
+        Ok(Self {
+            inner: paimon::table::Plan::from_splits(snapshot_id, splits),
+        })
+    }
+
     /// Snapshot selected by the scan, even when pruning produces no splits.
     /// `None` means no snapshot was selected, including snapshot-free format tables.
     fn snapshot_id(&self) -> Option<i64> {
@@ -1223,6 +1237,46 @@ impl PyPlan {
     }
 }
 
+/// Row-tracking routing metadata; constructing it performs no I/O.
+#[pyclass(name = "RowIdPlan", module = "pypaimon_rust.datafusion")]
+pub struct PyRowIdPlan {
+    inner: paimon::table::RowIdPlan,
+}
+
+#[pymethods]
+impl PyRowIdPlan {
+    #[new]
+    fn new(plan: &PyPlan) -> PyResult<Self> {
+        Ok(Self {
+            inner: paimon::table::RowIdPlan::from_plan(plan.inner.clone()).map_err(to_py_err)?,
+        })
+    }
+
+    fn snapshot_id(&self) -> Option<i64> {
+        self.inner.snapshot_id()
+    }
+
+    fn groups(&self) -> Vec<(i64, i64, PySplit)> {
+        self.inner
+            .groups()
+            .iter()
+            .map(|(first, last, split)| {
+                (
+                    *first,
+                    *last,
+                    PySplit {
+                        inner: split.clone(),
+                    },
+                )
+            })
+            .collect()
+    }
+
+    fn plan(&self) -> PyPlan {
+        self.inner.plan().clone().into()
+    }
+}
+
 #[pyclass(name = "Split", module = "pypaimon_rust.datafusion")]
 pub struct PySplit {
     pub(crate) inner: DataSplit,
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 299730b0f..e5688b9bd 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -616,12 +616,24 @@ fn arrow_upsert_batches(input: &Bound<'_, PyAny>) -> PyResult<Vec<RecordBatch>>
 
 struct UpdateContext {
     inner: TableUpdate,
+    planned: bool,
     table: Arc<Table>,
     table_location: String,
     commit_user: String,
 }
 
 impl UpdateContext {
+    // Reject a mismatched operation before consuming a Python input iterator
+    // or converting callbacks. Core independently enforces the same contract.
+    fn ensure_unplanned(&self) -> PyResult<()> {
+        if self.planned {
+            return Err(PyValueError::new_err(
+                "with_plan supports only row-ID update and delete operations",
+            ));
+        }
+        Ok(())
+    }
+
     fn with_blob_uri_reader_factory(
         &mut self,
         factory: Option<Py<PyAny>>,
@@ -634,6 +646,7 @@ impl UpdateContext {
 
     fn new(context: &WriteContext) -> PyResult<Self> {
         Ok(Self {
+            planned: false,
             inner: context
                 .table
                 .new_write_builder()
@@ -670,6 +683,7 @@ impl UpdateContext {
         when_matched: &Bound<'_, PyAny>,
         when_not_matched: &Bound<'_, PyAny>,
     ) -> PyResult<Vec<PyCommitMessage>> {
+        self.ensure_unplanned()?;
         let source = match source {
             Some(source) => match source.extract::<PyRef<crate::table::PyTable>>() {
                 Ok(table) => paimon::table::MergeSource::Table(table.inner.clone()),
@@ -706,6 +720,7 @@ impl UpdateContext {
         assignments: &Bound<'_, PyDict>,
         read_columns: Option<Vec<String>>,
     ) -> PyResult<Vec<PyCommitMessage>> {
+        self.ensure_unplanned()?;
         let predicate = predicate
             .map(|predicate| dict_to_table_predicate(predicate, self.table.schema(), true))
             .transpose()?;
@@ -795,6 +810,7 @@ impl UpdateContext {
         count: usize,
         read_type: Option<&str>,
     ) -> PyResult<crate::shard_update::PyShardTableUpdator> {
+        self.ensure_unplanned()?;
         let read_type = read_type
             .map(|json| {
                 serde_json::from_str::<RowType>(json)
@@ -829,6 +845,7 @@ impl UpdateContext {
         input: &Bound<'_, PyAny>,
         keys: Vec<String>,
     ) -> PyResult<Vec<PyCommitMessage>> {
+        self.ensure_unplanned()?;
         let batches = arrow_upsert_batches(input)?;
         let messages = py
             .detach(|| runtime().block_on(self.inner.upsert_by_arrow_with_key(batches, keys)))
@@ -845,6 +862,7 @@ impl UpdateContext {
         py: Python<'_>,
         predicate: Option<&Bound<'_, PyDict>>,
     ) -> PyResult<Vec<PyCommitMessage>> {
+        self.ensure_unplanned()?;
         let predicate = predicate
             .map(|predicate| dict_to_table_predicate(predicate, self.table.schema(), true))
             .transpose()?;
@@ -1064,6 +1082,18 @@ pub struct PyStreamTableUpdate {
 
 #[pymethods]
 impl PyStreamTableUpdate {
+    fn with_plan<'py>(
+        mut slf: PyRefMut<'py, Self>,
+        plan: &crate::read::PyPlan,
+    ) -> PyResult<PyRefMut<'py, Self>> {
+        slf.context
+            .inner
+            .with_plan(plan.inner.clone())
+            .map_err(to_py_err)?;
+        slf.context.planned = true;
+        Ok(slf)
+    }
+
     #[pyo3(signature = (factory))]
     fn with_blob_uri_reader_factory(
         mut slf: PyRefMut<'_, Self>,
@@ -1184,6 +1214,18 @@ impl PyStreamTableUpdate {
 
 #[pymethods]
 impl PyBatchTableUpdate {
+    fn with_plan<'py>(
+        mut slf: PyRefMut<'py, Self>,
+        plan: &crate::read::PyPlan,
+    ) -> PyResult<PyRefMut<'py, Self>> {
+        slf.context
+            .inner
+            .with_plan(plan.inner.clone())
+            .map_err(to_py_err)?;
+        slf.context.planned = true;
+        Ok(slf)
+    }
+
     #[pyo3(signature = (predicate))]
     fn delete_by_predicate(
         &self,
@@ -1615,6 +1657,27 @@ impl PyBatchTableCommit {
         .map_err(to_callback_error)
     }
 
+    /// Validate logical DML conflicts without publishing or deleting any files.
+    fn check_row_id_conflicts(
+        &self,
+        py: Python<'_>,
+        messages: &Bound<'_, PyAny>,
+        latest_snapshot_id: i64,
+        check_from_snapshot: i64,
+    ) -> PyResult<()> {
+        let messages = self
+            .context
+            .messages(messages, "check_row_id_conflicts", false)?;
+        py.detach(|| {
+            runtime().block_on(self.context.inner.check_row_id_conflicts(
+                &messages,
+                latest_snapshot_id,
+                check_from_snapshot,
+            ))
+        })
+        .map_err(to_py_err)
+    }
+
     fn truncate_table(&mut self, py: Python<'_>) -> PyResult<()> {
         self.check_committed()?;
         py.detach(|| runtime().block_on(self.context.inner.truncate_table()))
@@ -1784,6 +1847,186 @@ mod tests {
     use paimon::spec::{BigIntType, IntType, Schema, TableSchema, VarCharType};
     use pyo3::ffi::c_str;
 
+    #[test]
+    fn captured_plan_rejects_other_operations_before_python_input_consumption() {
+        Python::attach(|py| {
+            let schema = Schema::builder()
+                .column("id", DataType::Int(IntType::new()))
+                .option("row-tracking.enabled", "true")
+                .option("data-evolution.enabled", "true")
+                .build()
+                .unwrap();
+            let table = Arc::new(Table::new(
+                FileIOBuilder::new("memory").build().unwrap(),
+                Identifier::new("default", "planned_binding_boundary"),
+                "memory:/planned_binding_boundary".into(),
+                TableSchema::new(0, &schema),
+                None,
+            ));
+            let locals = PyDict::new(py);
+            locals
+                .set_item(
+                    "batch",
+                    Py::new(py, PyBatchWriteBuilder::new(table.clone())).unwrap(),
+                )
+                .unwrap();
+            locals
+                .set_item(
+                    "stream",
+                    Py::new(py, PyStreamWriteBuilder::new(table)).unwrap(),
+                )
+                .unwrap();
+            locals
+                .set_item("Plan", py.get_type::<crate::read::PyPlan>())
+                .unwrap();
+            py.run(c_str!(r#"
+class ExplodingInput:
+    def __iter__(self):
+        raise AssertionError('the supplied plan must reject input before iteration')
+
+for builder, identifier in [(batch, None), (stream, 7)]:
+    update = builder.new_update()
+    assert update.with_plan(Plan(None, [])) is update
+    assert update.with_plan(Plan(None, [])) is update
+    options = {} if identifier is None else {'commit_identifier': identifier}
+    operations = [
+        lambda: update.upsert_by_arrow_with_key(ExplodingInput(), ['id'], **options),
+        lambda: update.merge_into(ExplodingInput(), on=[], when_matched=[], when_not_matched=[], **options),
+    ]
+    for operation in operations:
+        try:
+            operation()
+        except ValueError as error:
+            assert 'with_plan supports only row-ID' in str(error), str(error)
+        else:
+            raise AssertionError('captured plan was silently ignored')
+"#),Some(&locals),Some(&locals)).unwrap();
+        });
+    }
+
+    #[test]
+    fn row_id_conflict_check_does_not_publish_or_consume_batch_commit() {
+        use arrow::array::{Array, Int32Array, Int64Array};
+        use futures::TryStreamExt;
+
+        Python::attach(|py| {
+            let schema = Schema::builder()
+                .column("id", DataType::Int(IntType::new()))
+                .option("row-tracking.enabled", "true")
+                .option("data-evolution.enabled", "true")
+                .build()
+                .unwrap();
+            let table = Arc::new(Table::new(
+                FileIOBuilder::new("memory").build().unwrap(),
+                Identifier::new("default", "conflict_check_binding"),
+                "memory:/conflict_check_binding".into(),
+                TableSchema::new(0, &schema),
+                None,
+            ));
+            let messages = py.detach(|| {
+                runtime().block_on(async {
+                    let mut writer = table.new_write_builder().new_write().unwrap();
+                    let batch = RecordBatch::try_new(
+                        paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap(),
+                        vec![Arc::new(Int32Array::from(vec![1, 2]))],
+                    )
+                    .unwrap();
+                    writer.write_arrow_batch(&batch).await.unwrap();
+                    table
+                        .new_write_builder()
+                        .new_commit()
+                        .commit(writer.prepare_commit().await.unwrap())
+                        .await
+                        .unwrap();
+                    writer.close().await;
+                    let updates = RecordBatch::try_from_iter(vec![
+                        (
+                            "_ROW_ID",
+                            Arc::new(Int64Array::from(vec![0])) as arrow::array::ArrayRef,
+                        ),
+                        (
+                            "id",
+                            Arc::new(Int32Array::from(vec![9])) as arrow::array::ArrayRef,
+                        ),
+                    ])
+                    .unwrap();
+                    table
+                        .new_write_builder()
+                        .new_update()
+                        .unwrap()
+                        .update_by_arrow_with_row_id(vec![updates])
+                        .await
+                        .unwrap()
+                })
+            });
+            let messages = messages
+                .into_iter()
+                .map(PyCommitMessage::new)
+                .collect::<Vec<_>>();
+            let builder = PyBatchWriteBuilder::new(table.clone());
+            let commit = Py::new(py, builder.new_commit().unwrap()).unwrap();
+            let locals = PyDict::new(py);
+            locals.set_item("commit", commit).unwrap();
+            locals.set_item("messages", messages).unwrap();
+            py.run(
+                c_str!("commit.check_row_id_conflicts(messages, 1, 1)"),
+                Some(&locals),
+                Some(&locals),
+            )
+            .unwrap();
+            let snapshot = py
+                .detach(|| runtime().block_on(table.snapshot_manager().get_latest_snapshot()))
+                .unwrap()
+                .unwrap();
+            assert_eq!(snapshot.id(), 1, "the independent check must not publish");
+            py.run(
+                c_str!("commit.commit(messages)"),
+                Some(&locals),
+                Some(&locals),
+            )
+            .unwrap();
+            py.detach(|| {
+                runtime().block_on(async {
+                    let snapshot = table
+                        .snapshot_manager()
+                        .get_latest_snapshot()
+                        .await
+                        .unwrap()
+                        .unwrap();
+                    assert_eq!(
+                        snapshot.id(),
+                        2,
+                        "the check must not consume the batch boundary"
+                    );
+                    let builder = table.new_read_builder();
+                    let plan = builder.new_scan().plan().await.unwrap();
+                    let batches = builder
+                        .new_read()
+                        .unwrap()
+                        .to_arrow(plan.splits())
+                        .unwrap()
+                        .try_collect::<Vec<_>>()
+                        .await
+                        .unwrap();
+                    let mut values = batches
+                        .iter()
+                        .flat_map(|batch| {
+                            let values = batch
+                                .column(0)
+                                .as_any()
+                                .downcast_ref::<Int32Array>()
+                                .unwrap();
+                            assert_eq!(values.null_count(), 0);
+                            values.values().to_vec()
+                        })
+                        .collect::<Vec<_>>();
+                    values.sort_unstable();
+                    assert_eq!(values, [2, 9]);
+                })
+            });
+        });
+    }
+
     #[test]
     fn public_uri_factory_and_commit_user_setters_are_chainable() {
         Python::attach(|py| {
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 01f877aa5..f4eec8267 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -31,6 +31,30 @@ pub enum Error {
     /// A configured resource budget could not admit the requested reservation.
     #[snafu(display("Paimon resource exhausted: {}", message))]
     ResourceExhausted { message: String },
+    /// A staged row-ID range no longer matches the current data-file layout.
+    ///
+    /// `current_row_count` is present only for a DELETE whose physical file
+    /// still exists with different row-ID metadata. Logical DML conflicts and
+    /// uncertain publication outcomes must not be classified as this error.
+    #[snafu(display(
+        "{}",
+        row_id_existence_conflict_message(
+            file_name,
+            first_row_id,
+            row_count,
+            bucket,
+            current_first_row_id,
+            current_row_count
+        )
+    ))]
+    RowIdExistenceConflict {
+        file_name: String,
+        first_row_id: Option<i64>,
+        row_count: i64,
+        bucket: i32,
+        current_first_row_id: Option<i64>,
+        current_row_count: Option<i64>,
+    },
     #[snafu(whatever, display("Paimon data invalid for {}: {:?}", message, source))]
     DataInvalid {
         message: String,
@@ -151,6 +175,34 @@ pub enum Error {
     },
 }
 
+fn row_id_existence_conflict_message(
+    file_name: &str,
+    first_row_id: &Option<i64>,
+    row_count: &i64,
+    bucket: &i32,
+    current_first_row_id: &Option<i64>,
+    current_row_count: &Option<i64>,
+) -> String {
+    let first_row_id = first_row_id.map_or_else(|| "null".into(), |value| value.to_string());
+    if let Some(current_row_count) = current_row_count {
+        let current_first_row_id =
+            current_first_row_id.map_or_else(|| "null".into(), |value| value.to_string());
+        format!(
+            "Row ID existence conflict: DELETE for file '{file_name}' references \
+             firstRowId={first_row_id}, rowCount={row_count} in bucket {bucket}, but the current \
+             file has firstRowId={current_first_row_id}, rowCount={current_row_count}. The file \
+             may have been reassigned by a concurrent commit."
+        )
+    } else {
+        format!(
+            "Row ID existence conflict: file '{file_name}' references firstRowId={first_row_id}, \
+             rowCount={row_count} in bucket {bucket}, but no matching file exists in the current \
+             snapshot. The referenced file may have been rewritten by a concurrent compaction \
+             or removed by an overwrite."
+        )
+    }
+}
+
 impl From<opendal::Error> for Error {
     fn from(source: opendal::Error) -> Self {
         Error::from_opendal_with_context(source, "IO operation failed on underlying storage")
diff --git a/crates/paimon/src/table/authorized_update_tests.rs b/crates/paimon/src/table/authorized_update_tests.rs
index eb24ea093..6de0fdc43 100644
--- a/crates/paimon/src/table/authorized_update_tests.rs
+++ b/crates/paimon/src/table/authorized_update_tests.rs
@@ -26,6 +26,190 @@ const EVOLUTION_OPTIONS: &[(&str, &str)] = &[
     ("deletion-vectors.enabled", "true"),
 ];
 
+#[tokio::test]
+async fn transported_row_id_plans_keep_the_rest_snapshot_after_concurrent_append() {
+    for delete in [false, true] {
+        let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+        fixture.append("seed", &[(1, "a", 10)], false).await;
+        fixture.state.deny_query.store(false, Ordering::SeqCst);
+        let plan = fixture
+            .table
+            .new_read_builder()
+            .new_scan()
+            .plan_all()
+            .await
+            .unwrap();
+        assert_eq!(plan.snapshot_id(), Some(1));
+        fixture.append("concurrent", &[(2, "a", 20)], false).await;
+        let requests = fixture.state.query_requests.load(Ordering::SeqCst);
+        let mut update = fixture.table.new_write_builder().new_update().unwrap();
+        update.with_plan(plan).unwrap();
+        let messages = if delete {
+            update.delete_by_row_id(vec![0]).await.unwrap()
+        } else {
+            update
+                .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+                .await
+                .unwrap()
+        };
+        assert_eq!(messages[0].check_from_snapshot, Some(1));
+        assert!(fixture.state.query_requests.load(Ordering::SeqCst) > requests);
+        fixture
+            .table
+            .new_write_builder()
+            .new_commit()
+            .commit(messages)
+            .await
+            .unwrap();
+        assert_eq!(
+            fixture.stored_rows().await,
+            if delete {
+                vec![(2, "a".into(), 20)]
+            } else {
+                vec![(1, "a".into(), 99), (2, "a".into(), 20)]
+            }
+        );
+        fixture
+            .table
+            .snapshot_manager()
+            .delete_snapshot(1)
+            .await
+            .unwrap();
+        let error = if delete {
+            update.delete_by_row_id(vec![0]).await.unwrap_err()
+        } else {
+            update
+                .update_by_arrow_with_row_id(vec![row_update(0, 100)])
+                .await
+                .unwrap_err()
+        };
+        assert!(matches!(
+            error,
+            crate::Error::SnapshotNotExist { snapshot_id: 1 }
+        ));
+    }
+}
+
+#[tokio::test]
+async fn transported_row_id_plans_refresh_whole_table_authorization_without_snapshot_files() {
+    for delete in [false, true] {
+        for response in [None]
+            .into_iter()
+            .chain(restricted_responses().into_iter().map(Some))
+        {
+            let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+            fixture.append("seed", &[(1, "a", 10)], false).await;
+            fixture.state.deny_query.store(false, Ordering::SeqCst);
+            let plan = fixture
+                .table
+                .new_read_builder()
+                .new_scan()
+                .plan_all()
+                .await
+                .unwrap();
+            let requests = fixture.state.query_requests.load(Ordering::SeqCst);
+            if let Some(response) = response {
+                *fixture.state.query_response.lock().await = response;
+            } else {
+                fixture.state.deny_query.store(true, Ordering::SeqCst);
+            }
+            let before = fixture
+                .table
+                .file_io()
+                .list_status_recursive(fixture.table.location())
+                .await
+                .unwrap()
+                .len();
+            let mut update = fixture.table.new_write_builder().new_update().unwrap();
+            update.with_plan(plan).unwrap();
+            let result = if delete {
+                update.delete_by_row_id(vec![0]).await
+            } else {
+                update
+                    .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+                    .await
+            };
+            assert!(
+                result.is_err(),
+                "captured grant must not survive a policy change"
+            );
+            assert!(fixture.state.query_requests.load(Ordering::SeqCst) > requests);
+            assert_eq!(
+                fixture
+                    .table
+                    .file_io()
+                    .list_status_recursive(fixture.table.location())
+                    .await
+                    .unwrap()
+                    .len(),
+                before
+            );
+        }
+    }
+    let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+    fixture.append("seed", &[(1, "a", 10)], false).await;
+    fixture.state.deny_query.store(false, Ordering::SeqCst);
+    let plan = fixture
+        .table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    fixture
+        .table
+        .file_io()
+        .delete_dir(&format!("{}/snapshot", fixture.table.location()))
+        .await
+        .unwrap();
+    assert!(fixture
+        .table
+        .snapshot_manager()
+        .get_snapshot(1)
+        .await
+        .is_err());
+    let mut update = fixture.table.new_write_builder().new_update().unwrap();
+    update.with_plan(plan).unwrap();
+    let messages = update
+        .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+        .await
+        .unwrap();
+    assert_eq!(messages[0].check_from_snapshot, Some(1));
+    assert_eq!(
+        update.delete_by_row_id(vec![0]).await.unwrap()[0].check_from_snapshot,
+        Some(1)
+    );
+}
+
+#[tokio::test]
+async fn restricted_read_plans_cannot_be_repurposed_as_mutation_plans_after_policy_relaxes() {
+    let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+    fixture.append("seed", &[(1, "a", 10)], false).await;
+    fixture.state.deny_query.store(false, Ordering::SeqCst);
+    *fixture.state.query_response.lock().await = restricted_responses().remove(0);
+    let plan = fixture
+        .table
+        .new_read_builder()
+        .new_scan()
+        .plan()
+        .await
+        .unwrap();
+    *fixture.state.query_response.lock().await = serde_json::json!({});
+    let mut update = fixture.table.new_write_builder().new_update().unwrap();
+    update.with_plan(plan).unwrap();
+    let error = update
+        .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("filtered or masked read plan"));
+    assert!(update
+        .delete_by_row_id(vec![0])
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("filtered or masked read plan"));
+}
+
 fn row_update(row_id: i64, value: i32) -> RecordBatch {
     RecordBatch::try_from_iter([
         (
diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs
index 2df6a7855..470f7aa02 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -39,7 +39,7 @@ use crate::table::index_file_path::IndexFileLocation;
 use crate::table::source::{data_evolution_anchor_file, is_data_evolution_normal_file};
 use crate::table::stats_filter::group_by_overlapping_row_id;
 use crate::table::Table;
-use crate::table::{DataSplit, RowRange};
+use crate::table::{DataSplit, Plan, RowRange};
 use crate::Result;
 use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StructArray};
 use arrow_buffer::NullBuffer;
@@ -355,6 +355,7 @@ impl DataEvolutionWriter {
         )?;
         let file_matches = group_matched_rows_by_file(&self.matched_batches, file_index)?;
         self.validate_partition_values(file_index, &file_matches)?;
+        index.validate_live_rows(&file_matches).await?;
 
         // 3. For each affected file: read original columns, apply updates, write partial files
         let mut writer = DataEvolutionPartialWriter::new(&self.table, self.update_columns.clone())?;
@@ -598,7 +599,15 @@ impl DataEvolutionDeleteWriter {
     }
 
     #[must_use = "commit messages must be passed to TableCommit"]
-    pub async fn prepare_commit(mut self) -> Result<Vec<CommitMessage>> {
+    pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> {
+        self.prepare_commit_inner(None).await
+    }
+
+    pub(super) async fn prepare_commit_with_plan(self, plan: &Plan) -> Result<Vec<CommitMessage>> {
+        self.prepare_commit_inner(Some(plan)).await
+    }
+
+    async fn prepare_commit_inner(mut self, captured: Option<&Plan>) -> Result<Vec<CommitMessage>> {
         dedup_i64_in_place(&mut self.row_ids);
         if self.row_ids.is_empty() {
             self.table
@@ -610,9 +619,18 @@ impl DataEvolutionDeleteWriter {
         }
 
         // A row filter or mask cannot authorize a physical file mutation.
-        let snapshot = super::time_travel::resolve_snapshot(&self.table).await?;
-        let scan_table = self.table.copy_with_pinned_snapshot(snapshot.as_ref());
-        let plan = scan_table.new_read_builder().new_scan().plan_all().await?;
+        let (snapshot, plan) = match captured {
+            Some(plan) => {
+                let (_, snapshot, plan) = authorize_row_id_plan(&self.table, plan).await?;
+                (snapshot, plan)
+            }
+            None => {
+                let snapshot = super::time_travel::resolve_snapshot(&self.table).await?;
+                let scan_table = self.table.copy_with_pinned_snapshot(snapshot.as_ref());
+                let plan = scan_table.new_read_builder().new_scan().plan_all().await?;
+                (snapshot, plan)
+            }
+        };
         let snapshot = snapshot.ok_or_else(|| crate::Error::DataInvalid {
             message: "No files with row tracking found in target table".into(),
             source: None,
@@ -1249,6 +1267,11 @@ pub(super) struct RowIdFileIndex {
 }
 
 impl RowIdFileIndex {
+    pub(super) async fn load_plan(table: &Table, plan: &Plan) -> Result<Self> {
+        let (read_table, _, plan) = authorize_row_id_plan(table, plan).await?;
+        Self::from_splits(read_table, plan.splits())
+    }
+
     pub(super) async fn load(table: &Table, snapshot_id: Option<i64>) -> Result<Self> {
         // Authorize before resolving an empty target. The original handle's
         // session is retained when the read view pins the selected snapshot.
@@ -1362,11 +1385,136 @@ impl RowIdFileIndex {
             .map(|&index| self.files[index].first_row_id)
             .collect())
     }
+
+    async fn validate_live_rows(&self, matches: &HashMap<usize, Vec<MatchedRow>>) -> Result<()> {
+        for (&file_pos, rows) in matches {
+            let group = &self.files[file_pos];
+            let anchor = data_evolution_anchor_file(&group.files)?;
+            let Some(deletion_file) = group.split.deletion_file_for_data_file(anchor) else {
+                continue;
+            };
+            let deletion_vector =
+                DeletionVectorFactory::read(self.read_table.file_io(), deletion_file).await?;
+            let first = anchor.first_row_id.expect("validated row-ID anchor");
+            for row in rows {
+                let row_id = group.first_row_id + row.offset as i64;
+                if deletion_vector.is_deleted((row_id - first) as u64) {
+                    return Err(crate::Error::DataInvalid {
+                        message: format!("Cannot update deleted _ROW_ID {row_id}"),
+                        source: None,
+                    });
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Validate transported planner metadata and mint a worker-local mutation grant.
+/// The caller supplies complete planner groups; manifests are deliberately not
+/// re-read to rediscover metadata already captured by the coordinator.
+async fn authorize_row_id_plan(
+    table: &Table,
+    plan: &Plan,
+) -> Result<(Table, Option<Snapshot>, Plan)> {
+    let required = table.schema().core_options().query_auth_enabled()
+        || plan.splits().iter().any(DataSplit::query_auth_required);
+    let grant = table.authorize_read(required, None).await?;
+    if required && grant.is_none() {
+        return Err(super::query_auth::unsupported(
+            "transported row-ID plans require REST authorization",
+        ));
+    }
+    if grant
+        .as_ref()
+        .is_some_and(|grant| !grant.is_unrestricted_whole_table())
+    {
+        return Err(super::query_auth::unsupported(
+            "row-ID updates require an unrestricted whole-table grant",
+        ));
+    }
+    super::RowIdPlan::from_plan(plan.clone())?;
+    if plan.splits().iter().any(|split| {
+        split
+            .query_auth_response()
+            .is_some_and(|response| !response.is_unrestricted())
+    }) {
+        return Err(super::query_auth::unsupported(
+            "a filtered or masked read plan cannot authorize row-ID mutations",
+        ));
+    }
+    let snapshot = match plan.snapshot_id() {
+        None => None,
+        Some(id) => {
+            let snapshot = if let Some(snapshot) = table.travel_snapshot() {
+                snapshot.clone()
+            } else if table.rest_env().is_some() {
+                match table.snapshot_manager().get_latest_snapshot().await? {
+                    Some(snapshot) if snapshot.id() == id => snapshot,
+                    // A concurrent append does not invalidate the captured
+                    // plan. Resolve its exact snapshot rather than replacing
+                    // the worker's conflict baseline with the latest one.
+                    _ => table.snapshot_manager().get_snapshot(id).await?,
+                }
+            } else {
+                table.snapshot_manager().get_snapshot(id).await?
+            };
+            if snapshot.id() != id || snapshot.schema_id() > table.schema().id() {
+                return Err(crate::Error::DataInvalid {
+                    message: "The row-ID plan does not match the table's captured snapshot/schema"
+                        .into(),
+                    source: None,
+                });
+            }
+            Some(snapshot)
+        }
+    };
+    let options = table.schema().core_options();
+    let computer = if table.schema().partition_keys().is_empty() {
+        None
+    } else {
+        Some(PartitionComputer::new(
+            table.schema().partition_keys(),
+            table.schema().fields(),
+            options.partition_default_name(),
+            options.legacy_partition_name(),
+        )?)
+    };
+    for split in plan.splits() {
+        let expected = bucket_path(
+            &table.data_file_location(),
+            computer.as_ref(),
+            split.partition(),
+            split.bucket(),
+        )?;
+        if split.bucket_path() != expected {
+            return Err(crate::Error::DataInvalid {
+                message: "The row-ID split bucket path does not belong to the target table".into(),
+                source: None,
+            });
+        }
+        for file in split.data_files() {
+            let (first, last) = file.row_id_range().expect("validated row-ID range");
+            if first < 0
+                || snapshot.as_ref().is_none_or(|snapshot| {
+                    file.schema_id > snapshot.schema_id()
+                        || snapshot.next_row_id().is_some_and(|next| last >= next)
+                })
+            {
+                return Err(crate::Error::DataInvalid {
+                    message: "The row-ID split contains file metadata outside its snapshot".into(),
+                    source: None,
+                });
+            }
+        }
+    }
+    let read_table = table.copy_with_pinned_snapshot(snapshot.as_ref());
+    Ok((read_table, snapshot, plan.clone().planned(grant)))
 }
 
 /// Dedicated column files can cross normal-file boundaries. They provide
 /// values, but must not merge anchors or introduce otherwise absent row IDs.
-fn group_by_normal_row_id(files: Vec<DataFileMeta>) -> Vec<Vec<DataFileMeta>> {
+pub(super) fn group_by_normal_row_id(files: Vec<DataFileMeta>) -> Vec<Vec<DataFileMeta>> {
     let (normal, dedicated): (Vec<_>, Vec<_>) =
         files.into_iter().partition(is_data_evolution_normal_file);
     group_by_overlapping_row_id(normal)
diff --git a/crates/paimon/src/table/global_index_build_builder.rs b/crates/paimon/src/table/global_index_build_builder.rs
index cb2036193..ab46595c5 100644
--- a/crates/paimon/src/table/global_index_build_builder.rs
+++ b/crates/paimon/src/table/global_index_build_builder.rs
@@ -21,6 +21,7 @@ use std::time::Instant;
 #[cfg(feature = "fulltext")]
 use super::full_text_index_build_builder::FullTextIndexBuildBuilder;
 use super::global_index_build_common::{resolve_index_fields, IndexColumns};
+use super::global_index_build_plan::{GlobalIndexBuildPlan, GlobalIndexBuildShard};
 use super::global_index_types::{
     normalize_global_index_type, normalize_queryable_global_index_type, BTREE_GLOBAL_INDEX_TYPE,
     FULL_TEXT_GLOBAL_INDEX_TYPE,
@@ -125,6 +126,47 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
         Ok(self.prepare().await?.messages)
     }
 
+    /// Capture independently executable vector-index shards without writing files.
+    ///
+    /// The plan owns its snapshot, schema and build options. Workers execute a
+    /// shard rather than replanning the table or publishing its commit messages.
+    pub async fn plan(&self) -> Result<GlobalIndexBuildPlan> {
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+        self.table.ensure_not_branch_reference_for_write()?;
+        let index_type = normalize_global_index_type(self.index_type.trim()).ok_or_else(|| {
+            Error::Unsupported {
+                message: format!("Unsupported global index type: {}", self.index_type),
+            }
+        })?;
+        if !crate::vindex::is_vindex_index_type(index_type) {
+            return Err(Error::Unsupported {
+                message: "Global index build plans currently support only vindex indexes".into(),
+            });
+        }
+        let columns = self.columns.as_ref().ok_or_else(|| Error::DataInvalid {
+            message: "Global index column is required".into(),
+            source: None,
+        })?;
+        let fields = resolve_index_fields(self.table, columns, index_type)?;
+        let mut builder = VindexIndexBuildBuilder::new(self.table, index_type);
+        builder
+            .with_index_column(fields[0].name())
+            .with_options(self.options.clone());
+        builder.partition_filter = self.partition_filter.clone();
+        builder.plan().await
+    }
+
+    /// Build only the files and physical row range captured by this shard.
+    /// Builder settings do not replace the plan's schema or index options.
+    pub async fn build_shard(&self, shard: &GlobalIndexBuildShard) -> Result<Vec<CommitMessage>> {
+        Ok(shard
+            .build(self.table)
+            .await?
+            .into_iter()
+            .map(|(message, _)| message)
+            .collect())
+    }
+
     /// Prepare and commit, rejecting changes to the snapshot used for the build.
     pub async fn execute(&self) -> Result<usize> {
         let prepared = self.prepare().await?;
diff --git a/crates/paimon/src/table/global_index_build_common/vector.rs b/crates/paimon/src/table/global_index_build_common/vector.rs
index a226597df..97ed5b71f 100644
--- a/crates/paimon/src/table/global_index_build_common/vector.rs
+++ b/crates/paimon/src/table/global_index_build_common/vector.rs
@@ -26,7 +26,7 @@ use crate::table::{RowRange, Table};
 use crate::{Error, Result};
 use std::collections::HashMap;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub(crate) struct VectorIndexShard {
     pub(crate) partition: BinaryRow,
     pub(crate) partition_bytes: Vec<u8>,
diff --git a/crates/paimon/src/table/global_index_build_plan.rs b/crates/paimon/src/table/global_index_build_plan.rs
new file mode 100644
index 000000000..9f8536d25
--- /dev/null
+++ b/crates/paimon/src/table/global_index_build_plan.rs
@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Snapshot-bound index tasks, independent of their scheduling framework.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use super::global_index_build_common::vector::VectorIndexShard;
+use super::vindex_index_build_builder::timing::VectorIndexBuildTiming;
+use super::vindex_index_build_builder::VindexIndexBuildBuilder;
+use super::{CommitMessage, RowRange, Table};
+use crate::spec::{Snapshot, TableSchema};
+use crate::{Error, Result};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct BuildContext {
+    location: String,
+    identifier: String,
+    branch: String,
+    schema: TableSchema,
+    snapshot: Option<Snapshot>,
+    column: String,
+    index_type: String,
+    options: HashMap<String, String>,
+}
+
+/// An immutable global-index build over one captured table snapshot.
+/// Contains metadata only: clients, FileIO and runtime handles stay in Table.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GlobalIndexBuildPlan {
+    context: BuildContext,
+    shards: Vec<VectorIndexShard>,
+}
+
+impl GlobalIndexBuildPlan {
+    pub(crate) fn new(
+        table: &Table,
+        snapshot: Option<Snapshot>,
+        column: String,
+        index_type: String,
+        options: HashMap<String, String>,
+        shards: Vec<VectorIndexShard>,
+    ) -> Self {
+        Self {
+            context: BuildContext {
+                location: table.location().to_string(),
+                identifier: table.identifier().full_name(),
+                branch: table.branch().to_string(),
+                schema: table.schema().clone(),
+                snapshot,
+                column,
+                index_type,
+                options,
+            },
+            shards,
+        }
+    }
+
+    pub fn snapshot_id(&self) -> Option<i64> {
+        self.context.snapshot.as_ref().map(Snapshot::id)
+    }
+
+    /// Each task carries the same captured schema, options and snapshot.
+    pub fn shards(&self) -> Vec<GlobalIndexBuildShard> {
+        self.shards
+            .iter()
+            .cloned()
+            .map(|shard| GlobalIndexBuildShard {
+                context: self.context.clone(),
+                shard,
+            })
+            .collect()
+    }
+}
+
+/// A complete physical row range and its files, executable on another worker.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GlobalIndexBuildShard {
+    context: BuildContext,
+    shard: VectorIndexShard,
+}
+
+impl GlobalIndexBuildShard {
+    pub fn snapshot_id(&self) -> i64 {
+        self.shard.snapshot_id
+    }
+
+    pub fn row_range(&self) -> RowRange {
+        RowRange::new(self.shard.row_range_start, self.shard.row_range_end)
+    }
+
+    pub(in crate::table) async fn build(
+        &self,
+        table: &Table,
+    ) -> Result<Option<(CommitMessage, Option<VectorIndexBuildTiming>)>> {
+        // A serialized task grants no authorization. Check the worker's table
+        // before replacing its schema with the captured read view.
+        table.schema().core_options().ensure_read_authorized()?;
+        table.ensure_not_branch_reference_for_write()?;
+        if table.location() != self.context.location
+            || table.identifier().full_name() != self.context.identifier
+            || table.branch() != self.context.branch
+        {
+            return Err(Error::DataInvalid {
+                message: "Global index build shard belongs to a different table or branch".into(),
+                source: None,
+            });
+        }
+        let snapshot = self
+            .context
+            .snapshot
+            .as_ref()
+            .ok_or_else(|| Error::DataInvalid {
+                message: "Global index build shard requires a captured snapshot".into(),
+                source: None,
+            })?;
+        if snapshot.id() != self.shard.snapshot_id
+            || self.shard.row_range_start < 0
+            || self.shard.row_range_end < self.shard.row_range_start
+        {
+            return Err(Error::DataInvalid {
+                message: "Invalid global index build shard snapshot or row range".into(),
+                source: None,
+            });
+        }
+        let table = table
+            .copy_with_resolved_schema(self.context.schema.clone(), &self.context.branch)?
+            .copy_with_pinned_snapshot(Some(snapshot));
+        let mut builder = VindexIndexBuildBuilder::new(&table, &self.context.index_type);
+        builder
+            .with_index_column(&self.context.column)
+            .with_options(self.context.options.clone());
+        builder.build_planned_shard(&self.shard).await
+    }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 87e89f56e..1f95c5af6 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -79,6 +79,7 @@ mod full_text_index_build_builder;
 mod full_text_search_builder;
 mod global_index_build_builder;
 pub(crate) mod global_index_build_common;
+mod global_index_build_plan;
 mod global_index_drop_builder;
 mod global_index_freshness;
 mod global_index_result;
@@ -222,6 +223,7 @@ pub use full_text_search_builder::{
 };
 use futures::stream::BoxStream;
 pub use global_index_build_builder::GlobalIndexBuildBuilder;
+pub use global_index_build_plan::{GlobalIndexBuildPlan, GlobalIndexBuildShard};
 pub use global_index_drop_builder::GlobalIndexDropBuilder;
 pub use global_index_result::{
     GlobalIndexResult, GlobalIndexResultReadBuilder, PrimaryKeySearchPosition,
@@ -263,7 +265,7 @@ pub use table_merge_into::{
 pub use table_read::{AuditLogRead, TableRead};
 pub use table_scan::{AuditLogScan, FileScanPlan, TableScan};
 pub use table_update::TableUpdate;
-pub use table_update_by_row_id::TableUpdateByRowId;
+pub use table_update_by_row_id::{RowIdPlan, TableUpdateByRowId};
 pub use table_write::TableWrite;
 pub use tag_manager::TagManager;
 pub use update_assignment::UpdateAssignment;
diff --git a/crates/paimon/src/table/source.rs b/crates/paimon/src/table/source.rs
index 2becb42f4..01b292aa1 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -1672,7 +1672,7 @@ impl Default for DataSplitBuilder {
 /// Read plan: splits and the snapshot they were planned from.
 ///
 /// Reference: [org.apache.paimon.table.source.PlanImpl](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/PlanImpl.java)
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Plan {
     splits: Vec<DataSplit>,
     snapshot_id: Option<i64>,
@@ -1680,6 +1680,16 @@ pub struct Plan {
 }
 
 impl Plan {
+    /// Restore a captured plan from its snapshot and typed splits.
+    ///
+    /// This transports planning metadata; it does not authorize access to the
+    /// referenced files. Readers and mutation workers authorize independently.
+    pub fn from_splits(snapshot_id: Option<i64>, splits: Vec<DataSplit>) -> Self {
+        let mut plan = Self::new(splits);
+        plan.snapshot_id = snapshot_id;
+        plan
+    }
+
     pub fn new(splits: Vec<DataSplit>) -> Self {
         let mut plan = Self {
             splits,
diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs
index 878793679..72e7ec2ba 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -289,6 +289,70 @@ impl TableCommit {
         self
     }
 
+    /// Check logical row-ID/column conflicts before rebasing staged updates.
+    ///
+    /// Both snapshot IDs are resolved through this table's snapshot manager.
+    /// The captured interval must not cross a schema change, and this
+    /// committer's resolved schema must still be current. A newer schema may
+    /// exist before the next data snapshot is published.
+    /// Like Java's `checkRowIdFromSnapshot(..., false)`, COMPACT snapshots are
+    /// excluded: the caller may rewrite physical layouts only after this check
+    /// has rejected concurrent logical writes. This method publishes nothing
+    /// and never deletes prepared files, including on an uncertain outcome.
+    pub async fn check_row_id_conflicts(
+        &self,
+        messages: &[CommitMessage],
+        latest_snapshot_id: i64,
+        check_from_snapshot: i64,
+    ) -> Result<()> {
+        if check_from_snapshot <= 0 || check_from_snapshot > latest_snapshot_id {
+            return Err(crate::Error::DataInvalid {
+                message:
+                    "Row-ID conflict checks require 0 < check-from snapshot <= latest snapshot"
+                        .into(),
+                source: None,
+            });
+        }
+        if Self::check_from_snapshot(messages)? != Some(check_from_snapshot) {
+            return Err(crate::Error::DataInvalid {
+                message: "Row-ID conflict check baseline must match the commit messages".into(),
+                source: None,
+            });
+        }
+        super::write_builder::ensure_table_write_authorized(&self.table)?;
+        if !self.data_evolution_enabled {
+            return Err(crate::Error::Unsupported {
+                message: "Row-ID conflict checks require a data-evolution table".into(),
+            });
+        }
+        reject_compact_increment(messages)?;
+        let latest = self
+            .snapshot_manager
+            .get_snapshot(latest_snapshot_id)
+            .await?;
+        let baseline = if check_from_snapshot == latest_snapshot_id {
+            latest.clone()
+        } else {
+            self.snapshot_manager
+                .get_snapshot(check_from_snapshot)
+                .await?
+        };
+        if baseline.schema_id() != latest.schema_id()
+            || self.latest_schema_id().await? != self.table.schema().id()
+        {
+            return Err(crate::Error::DataInvalid {
+                message: "Row-ID conflict checks require an unchanged, current table schema".into(),
+                source: None,
+            });
+        }
+        self.check_row_id_from_snapshot(
+            Some(&latest),
+            &self.messages_to_entries(messages),
+            Some(check_from_snapshot),
+        )
+        .await
+    }
+
     /// Attach caller properties to snapshots published by this committer.
     ///
     /// Like Java's ManifestCommittable properties, these describe the current
@@ -2756,6 +2820,32 @@ impl TableCommit {
         delta_entries: &[ManifestEntry],
         next_row_id: Option<i64>,
     ) -> Result<()> {
+        // File identity deliberately excludes row IDs. Check DELETE metadata
+        // before identifier-based merging can cancel a reassigned current ADD.
+        let deleted = delta_entries
+            .iter()
+            .filter(|entry| *entry.kind() == FileKind::Delete)
+            .map(|entry| (commit_file_identifier(entry), entry))
+            .collect::<HashMap<_, _>>();
+        for current in base_entries
+            .iter()
+            .filter(|entry| *entry.kind() == FileKind::Add)
+        {
+            if let Some(deleted) = deleted.get(&commit_file_identifier(current)) {
+                if current.file().first_row_id != deleted.file().first_row_id
+                    || current.file().row_count != deleted.file().row_count
+                {
+                    return Err(crate::Error::RowIdExistenceConflict {
+                        file_name: deleted.file().file_name.clone(),
+                        first_row_id: deleted.file().first_row_id,
+                        row_count: deleted.file().row_count,
+                        bucket: deleted.bucket(),
+                        current_first_row_id: current.file().first_row_id,
+                        current_row_count: Some(current.file().row_count),
+                    });
+                }
+            }
+        }
         let Some(next_row_id) = next_row_id else {
             return Ok(());
         };
@@ -2818,15 +2908,13 @@ impl TableCommit {
                 entry.file().row_count,
             );
             if !existing_index.contains(&key) {
-                return Err(crate::Error::DataInvalid {
-                    message: format!(
-                        "Row ID existence conflict: file '{}' references first_row_id={}, row_count={} in bucket {}, but no matching file exists in the current snapshot.",
-                        entry.file().file_name,
-                        first_row_id,
-                        entry.file().row_count,
-                        entry.bucket(),
-                    ),
-                    source: None,
+                return Err(crate::Error::RowIdExistenceConflict {
+                    file_name: entry.file().file_name.clone(),
+                    first_row_id: Some(first_row_id),
+                    row_count: entry.file().row_count,
+                    bucket: entry.bucket(),
+                    current_first_row_id: None,
+                    current_row_count: None,
                 });
             }
         }
@@ -3711,6 +3799,7 @@ mod tests {
         include!("table_commit/property_tests.rs");
         include!("table_commit/callback_tests.rs");
         include!("table_commit/index_delete_tests.rs");
+        include!("table_commit/row_id_conflict_tests.rs");
     }
 
     #[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs
new file mode 100644
index 000000000..99e90a4f1
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs
@@ -0,0 +1,222 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#[test]
+fn row_id_delete_must_match_the_current_file_layout() {
+    let commit = setup_data_evolution_commit(&test_file_io(), "memory:/row_id_delete_layout");
+    let mut file = test_data_file("data-reassigned.parquet", 10);
+    file.first_row_id = Some(10);
+    let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1, file.clone(), 2);
+    file.first_row_id = Some(0);
+    let stale_delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, file, 2);
+    let error = commit
+        .check_row_id_existence(&[current], &[stale_delete], Some(20))
+        .expect_err("a DELETE must not cancel the current file after row-ID reassignment");
+    assert!(error.to_string().contains("Row ID existence conflict"));
+    assert!(matches!(
+        error,
+        crate::Error::RowIdExistenceConflict {
+            first_row_id: Some(0),
+            row_count: 10,
+            bucket: 0,
+            current_first_row_id: Some(10),
+            current_row_count: Some(10),
+            ..
+        }
+    ));
+}
+
+#[test]
+fn row_id_existence_preserves_missing_range_details() {
+    let commit = setup_data_evolution_commit(&test_file_io(), "memory:/missing_row_id_range");
+    let mut file = test_data_file("opaque '(4): file.parquet", 4);
+    file.first_row_id = Some(7);
+    let entry = ManifestEntry::new(FileKind::Add, vec![], 3, -1, file, 2);
+    let error = commit
+        .check_row_id_existence(&[], &[entry], Some(20))
+        .unwrap_err();
+    assert!(matches!(error, crate::Error::RowIdExistenceConflict {
+        ref file_name, first_row_id: Some(7), row_count: 4, bucket: 3,
+        current_first_row_id: None, current_row_count: None
+    } if file_name == "opaque '(4): file.parquet"));
+}
+
+#[test]
+fn row_id_delete_metadata_check_does_not_require_next_row_id() {
+    let commit = setup_data_evolution_commit(&test_file_io(), "memory:/delete_without_counter");
+    for (current_id, current_count) in [(None, 10), (Some(0), 11)] {
+        let mut file = test_data_file("same.parquet", current_count);
+        file.first_row_id = current_id;
+        let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1, file, 2);
+        let mut old = test_data_file("same.parquet", 10);
+        old.first_row_id = Some(0);
+        let old = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, old, 2);
+        assert!(
+            matches!(commit.check_row_id_existence(&[current], &[old], None),
+            Err(crate::Error::RowIdExistenceConflict { current_first_row_id, current_row_count: Some(count), .. })
+            if current_first_row_id == current_id && count == current_count)
+        );
+    }
+    let file = test_data_file("ordinary.parquet", 10);
+    let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1, file.clone(), 2);
+    let delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, file, 2);
+    commit
+        .check_row_id_existence(&[current], &[delete], None)
+        .unwrap();
+    commit.check_row_id_existence(&[], &[], None).unwrap();
+}
+
+fn row_id_check_message(name: &str, column: &str) -> CommitMessage {
+    let mut file = test_data_file(name, 10);
+    file.first_row_id = Some(0);
+    file.file_source = Some(0);
+    file.write_cols = Some(vec![column.into()]);
+    let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0, vec![file]);
+    message.check_from_snapshot = Some(1);
+    message
+}
+
+async fn row_id_check_history(path: &str) -> TableCommit {
+    let io = test_file_io();
+    setup_dirs(&io, path).await;
+    let commit = setup_data_evolution_commit(&io, path);
+    let mut file = test_data_file("base.parquet", 10);
+    file.file_source = Some(0);
+    commit
+        .commit_with_identifier(
+            vec![CommitMessage::new(
+                EMPTY_SERIALIZED_ROW.clone(),
+                0,
+                vec![file],
+            )],
+            1,
+        )
+        .await
+        .unwrap();
+    commit
+        .commit_with_identifier(vec![row_id_check_message("concurrent.parquet", "name")], 2)
+        .await
+        .unwrap();
+    commit
+}
+
+#[tokio::test]
+async fn standalone_row_id_check_rejects_logical_changes_without_deleting_staged_files() {
+    let commit = row_id_check_history("memory:/standalone_row_id_check").await;
+    let io = commit.table.file_io();
+    let path = "memory:/standalone_row_id_check/bucket-0/staged.parquet";
+    io.new_output(path)
+        .unwrap()
+        .write(bytes::Bytes::from_static(b"prepared"))
+        .await
+        .unwrap();
+    let error = commit
+        .check_row_id_conflicts(&[row_id_check_message("staged.parquet", "name")], 2, 1)
+        .await
+        .unwrap_err();
+    assert!(matches!(error, crate::Error::DataInvalid { .. }));
+    assert!(error.to_string().contains("multiple MERGE INTO"));
+    assert!(
+        io.exists(path).await.unwrap(),
+        "a failed check must retain prepared files"
+    );
+    assert_eq!(
+        commit
+            .snapshot_manager
+            .get_latest_snapshot_id()
+            .await
+            .unwrap(),
+        Some(2)
+    );
+    commit
+        .check_row_id_conflicts(&[row_id_check_message("id-only.parquet", "id")], 2, 1)
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn standalone_row_id_check_skips_compaction_history() {
+    let commit = row_id_check_history("memory:/standalone_compact_history").await;
+    // Replace only the fixture's commit kind. The same overlapping delta must
+    // be ignored when it describes physical compaction rather than a DML write.
+    let snapshot = commit.snapshot_manager.get_snapshot(2).await.unwrap();
+    let mut json = serde_json::to_value(snapshot).unwrap();
+    json["commitKind"] = "COMPACT".into();
+    let path = format!("{}/snapshot-2", commit.snapshot_manager.snapshot_dir());
+    commit
+        .table
+        .file_io()
+        .new_output(&path)
+        .unwrap()
+        .write(serde_json::to_vec(&json).unwrap().into())
+        .await
+        .unwrap();
+    commit
+        .check_row_id_conflicts(&[row_id_check_message("staged.parquet", "name")], 2, 1)
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn standalone_row_id_check_validates_snapshot_baselines_and_current_schema() {
+    let commit = row_id_check_history("memory:/standalone_check_validation").await;
+    let message = row_id_check_message("staged.parquet", "name");
+    for (latest, baseline) in [(2, 0), (1, 2), (0, 1), (2, -1), (2, 2)] {
+        assert!(matches!(
+            commit
+                .check_row_id_conflicts(std::slice::from_ref(&message), latest, baseline)
+                .await,
+            Err(crate::Error::DataInvalid { .. })
+        ));
+    }
+    assert!(matches!(
+        commit.check_row_id_conflicts(&[], 2, 1).await,
+        Err(crate::Error::DataInvalid { .. })
+    ));
+    assert!(commit
+        .check_row_id_conflicts(std::slice::from_ref(&message), 3, 1)
+        .await
+        .is_err());
+    save_schema(
+        commit.table.file_io(),
+        "memory:/standalone_check_validation",
+        1,
+    )
+    .await;
+    let error = commit
+        .check_row_id_conflicts(&[message], 2, 1)
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("current table schema"));
+}
+
+#[tokio::test]
+async fn standalone_row_id_check_fails_closed_for_missing_baseline_and_assembled_authorization() {
+    let commit = row_id_check_history("memory:/standalone_missing_baseline").await;
+    commit.snapshot_manager.delete_snapshot(1).await.unwrap();
+    assert!(commit
+        .check_row_id_conflicts(&[row_id_check_message("staged.parquet", "name")], 2, 1)
+        .await
+        .is_err());
+    let guarded = TableCommit::new(crate::table::query_auth_table(), "guarded".into());
+    let error = guarded
+        .check_row_id_conflicts(&[row_id_check_message("staged.parquet", "name")], 2, 1)
+        .await
+        .unwrap_err();
+    assert!(matches!(error, crate::Error::Unsupported { .. }));
+    assert!(error.to_string().contains("query-auth.enabled"));
+}
diff --git a/crates/paimon/src/table/table_update.rs b/crates/paimon/src/table/table_update.rs
index 1eac40635..d9257a483 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -56,6 +56,7 @@ pub struct TableUpdate {
     table: Table,
     commit_user: String,
     update_cols: Option<Vec<String>>,
+    plan: Option<super::Plan>,
     blob_uri_reader_factory: Option<std::sync::Arc<dyn crate::io::UriReaderFactory>>,
 }
 
@@ -65,10 +66,39 @@ impl TableUpdate {
             table: table.clone(),
             commit_user,
             update_cols: None,
+            plan: None,
             blob_uri_reader_factory: None,
         }
     }
 
+    /// Reuse a captured `plan_all` result for row-ID updates and deletes.
+    ///
+    /// Each split must contain complete logical file groups, including their
+    /// partial-column files and deletion vectors. Splits may be distributed
+    /// among workers; the supplied row IDs must belong to that worker's groups.
+    /// Arbitrarily trimmed or incremental splits are not mutation plans.
+    /// Workers authorize the whole table again and retain the captured snapshot
+    /// as the conflict baseline without replanning its data manifests. Deletes
+    /// still read the captured index manifest to preserve all deletion vectors
+    /// when replacing a bucket's shared index file. Assign all deletes for one
+    /// partition/bucket to the same worker.
+    /// Predicate, key-based, MERGE and shard operations reject this setting;
+    /// use a separately configured updater for those operations.
+    pub fn with_plan(&mut self, plan: super::Plan) -> crate::Result<&mut Self> {
+        super::RowIdPlan::from_plan(plan.clone())?;
+        self.plan = Some(plan);
+        Ok(self)
+    }
+
+    fn ensure_unplanned(&self) -> crate::Result<()> {
+        if self.plan.is_some() {
+            return Err(crate::Error::Unsupported {
+                message: "with_plan supports only row-ID update and delete operations".into(),
+            });
+        }
+        Ok(())
+    }
+
     /// Configure readers for Blob input references. The same factory supplies
     /// matched updates and appended rows in upsert and MERGE operations.
     pub fn with_blob_uri_reader_factory(
@@ -120,6 +150,7 @@ impl TableUpdate {
         shard_count: usize,
         read_type: Option<Vec<DataField>>,
     ) -> crate::Result<super::ShardTableUpdater> {
+        self.ensure_unplanned()?;
         let mut updater = super::ShardTableUpdater::new(
             &self.table,
             read_type,
@@ -170,7 +201,10 @@ impl TableUpdate {
 
     /// Create a row-ID updater sharing one snapshot across per-call columns.
     pub async fn new_update_by_row_id(&self) -> crate::Result<TableUpdateByRowId> {
-        let mut writer = TableUpdateByRowId::new(&self.table).await?;
+        let mut writer = match &self.plan {
+            Some(plan) => TableUpdateByRowId::with_plan(&self.table, plan).await?,
+            None => TableUpdateByRowId::new(&self.table).await?,
+        };
         writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
         Ok(writer)
     }
@@ -233,6 +267,7 @@ impl TableUpdate {
         assignments: Vec<(String, super::UpdateAssignment)>,
         read_columns: Vec<String>,
     ) -> crate::Result<Vec<CommitMessage>> {
+        self.ensure_unplanned()?;
         super::table_update_predicate::update(
             &self.table,
             predicate,
@@ -261,6 +296,7 @@ impl TableUpdate {
         batches: Vec<RecordBatch>,
         upsert_keys: Vec<String>,
     ) -> crate::Result<Vec<CommitMessage>> {
+        self.ensure_unplanned()?;
         let mut writer = super::table_upsert::TableUpsert::new(
             &self.table,
             self.commit_user.clone(),
@@ -286,6 +322,7 @@ impl TableUpdate {
         when_matched: Vec<super::WhenMatched>,
         when_not_matched: Vec<super::WhenNotMatched>,
     ) -> crate::Result<Vec<CommitMessage>> {
+        self.ensure_unplanned()?;
         super::table_merge_into::merge_into(
             &self.table,
             &self.commit_user,
@@ -311,6 +348,7 @@ impl TableUpdate {
         &self,
         predicate: Option<crate::spec::Predicate>,
     ) -> crate::Result<Vec<CommitMessage>> {
+        self.ensure_unplanned()?;
         super::table_update_delete::delete(&self.table, predicate).await
     }
 
@@ -323,10 +361,17 @@ impl TableUpdate {
         // ordinary write builder continues rejecting historical handles.
         let mut writer = super::DataEvolutionDeleteWriter::new(&self.table)?;
         writer.add_row_ids(row_ids)?;
-        writer.prepare_commit().await
+        match &self.plan {
+            Some(plan) => writer.prepare_commit_with_plan(plan).await,
+            None => writer.prepare_commit().await,
+        }
     }
 }
 
 fn same_columns(left: &[String], right: &[String]) -> bool {
     left.len() == right.len() && left.iter().all(|column| right.contains(column))
 }
+
+#[cfg(test)]
+#[path = "table_update_plan_tests.rs"]
+mod plan_tests;
diff --git a/crates/paimon/src/table/table_update_by_row_id.rs b/crates/paimon/src/table/table_update_by_row_id.rs
index afec5f00d..acf5eba52 100644
--- a/crates/paimon/src/table/table_update_by_row_id.rs
+++ b/crates/paimon/src/table/table_update_by_row_id.rs
@@ -17,8 +17,8 @@
 
 //! Row-ID updates sharing one target snapshot and file index.
 
-use super::data_evolution_writer::RowIdFileIndex;
-use crate::table::{CommitMessage, DataEvolutionWriter, Table};
+use super::data_evolution_writer::{group_by_normal_row_id, RowIdFileIndex};
+use crate::table::{CommitMessage, DataEvolutionWriter, DataSplit, Plan, Table};
 use arrow_array::RecordBatch;
 use std::collections::{HashMap, HashSet};
 
@@ -29,6 +29,99 @@ fn invalid(message: impl Into<String>) -> crate::Error {
     }
 }
 
+/// Routing metadata for complete row-tracked file groups in a captured plan.
+///
+/// Constructing this object performs no I/O and grants no file access. The
+/// reader or mutation executing the retained plan must authorize independently.
+#[derive(Debug, Clone)]
+pub struct RowIdPlan {
+    plan: Plan,
+    groups: Vec<(i64, i64, DataSplit)>,
+}
+
+impl RowIdPlan {
+    pub fn from_plan(plan: Plan) -> crate::Result<Self> {
+        if plan.snapshot_id().is_none() && !plan.splits().is_empty() {
+            return Err(invalid("A row-ID plan requires its captured snapshot ID"));
+        }
+        let mut groups = Vec::new();
+        for split in plan.splits() {
+            if Some(split.snapshot_id()) != plan.snapshot_id()
+                || split.is_streaming()
+                || split.row_ranges().is_some()
+                || split.scores().is_some()
+            {
+                return Err(invalid(
+                    "A row-ID plan requires full snapshot file groups without row selection",
+                ));
+            }
+            if split.data_files().is_empty() {
+                return Err(invalid("A row-ID split must contain data files"));
+            }
+            if split
+                .data_deletion_files()
+                .is_some_and(|files| files.len() != split.data_files().len())
+            {
+                return Err(invalid(
+                    "Row-ID deletion metadata must match the captured data files",
+                ));
+            }
+            let mut names = HashSet::new();
+            for file in split.data_files() {
+                if file.row_id_range().is_none() || !names.insert(&file.file_name) {
+                    return Err(invalid(
+                        "A row-ID split contains invalid or duplicate file metadata",
+                    ));
+                }
+            }
+            let file_groups = group_by_normal_row_id(split.data_files().to_vec());
+            if file_groups.is_empty() {
+                return Err(invalid("A row-ID split requires a normal anchor file"));
+            }
+            let mut grouped = HashSet::new();
+            for files in file_groups {
+                let anchor = super::source::data_evolution_anchor_file(&files)?;
+                let (first, last) = anchor
+                    .row_id_range()
+                    .ok_or_else(|| invalid("Invalid row-ID anchor range"))?;
+                for file in &files {
+                    grouped.insert(file.file_name.clone());
+                    if super::source::is_data_evolution_normal_file(file)
+                        && file.row_id_range() != Some((first, last))
+                    {
+                        return Err(invalid(
+                            "Overlapping row-ID anchors must have identical ranges",
+                        ));
+                    }
+                }
+                groups.push((first, last, split.for_data_evolution_files(files)));
+            }
+            if grouped.len() != names.len() {
+                return Err(invalid(
+                    "A row-ID split contains a dedicated file without its normal anchor",
+                ));
+            }
+        }
+        groups.sort_by_key(|(first, _, _)| *first);
+        if groups.windows(2).any(|pair| pair[0].1 >= pair[1].0) {
+            return Err(invalid("A row-ID plan contains overlapping file groups"));
+        }
+        Ok(Self { plan, groups })
+    }
+
+    pub fn snapshot_id(&self) -> Option<i64> {
+        self.plan.snapshot_id()
+    }
+
+    pub fn groups(&self) -> &[(i64, i64, DataSplit)] {
+        &self.groups
+    }
+
+    pub fn plan(&self) -> &Plan {
+        &self.plan
+    }
+}
+
 /// Update columns selected per input table, as in PyPaimon.
 ///
 /// Each call to `update_columns` stages its files immediately and returns all
@@ -47,6 +140,10 @@ impl TableUpdateByRowId {
         Self::with_index(table, RowIdFileIndex::load(table, None).await?)
     }
 
+    pub(super) async fn with_plan(table: &Table, plan: &Plan) -> crate::Result<Self> {
+        Self::with_index(table, RowIdFileIndex::load_plan(table, plan).await?)
+    }
+
     pub(super) fn with_index(table: &Table, index: RowIdFileIndex) -> crate::Result<Self> {
         // Validate table layout without fixing any update columns.
         let _ = DataEvolutionWriter::new(table, Vec::new())?;
diff --git a/crates/paimon/src/table/table_update_plan_tests.rs b/crates/paimon/src/table/table_update_plan_tests.rs
new file mode 100644
index 000000000..b4fb6dddb
--- /dev/null
+++ b/crates/paimon/src/table/table_update_plan_tests.rs
@@ -0,0 +1,418 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::*;
+use crate::catalog::Identifier;
+use crate::io::FileIOBuilder;
+use crate::spec::{DataType, IntType, Schema, TableSchema};
+use crate::table::{DataSplitBuilder, Plan, RowRange};
+use arrow_array::{ArrayRef, Int32Array, Int64Array};
+use std::sync::Arc;
+
+async fn table(name: &str) -> Table {
+    let schema = Schema::builder()
+        .column("value", DataType::Int(IntType::new()))
+        .option("bucket", "-1")
+        .option("data-evolution.enabled", "true")
+        .option("row-tracking.enabled", "true")
+        .option("deletion-vectors.enabled", "true")
+        .build()
+        .unwrap();
+    let table = Table::new(
+        FileIOBuilder::new("memory").build().unwrap(),
+        Identifier::new("default", name),
+        format!("memory:/{name}"),
+        TableSchema::new(0, &schema),
+        None,
+    );
+    for value in [10, 20] {
+        append(&table, value).await;
+    }
+    table
+}
+
+async fn append(table: &Table, value: i32) {
+    let batch = RecordBatch::try_from_iter(vec![(
+        "value",
+        Arc::new(Int32Array::from(vec![value])) as ArrayRef,
+    )])
+    .unwrap();
+    let builder = table.new_write_builder();
+    let mut writer = builder.new_write().unwrap();
+    writer.write_arrow_batch(&batch).await.unwrap();
+    builder
+        .new_commit()
+        .commit(writer.prepare_commit().await.unwrap())
+        .await
+        .unwrap();
+}
+
+fn update_batch(row_id: i64) -> RecordBatch {
+    RecordBatch::try_from_iter(vec![
+        (
+            "_ROW_ID",
+            Arc::new(Int64Array::from(vec![row_id])) as ArrayRef,
+        ),
+        ("value", Arc::new(Int32Array::from(vec![99])) as ArrayRef),
+    ])
+    .unwrap()
+}
+
+#[tokio::test]
+async fn planned_updates_and_deletes_keep_the_captured_baseline_without_manifest_rescan() {
+    let table = table("planned_mutation_baseline").await;
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    assert_eq!(plan.snapshot_id(), Some(2));
+    append(&table, 30).await;
+    table
+        .file_io()
+        .delete_dir(&format!("{}/manifest", table.location()))
+        .await
+        .unwrap();
+
+    let mut update = table.new_write_builder().new_update().unwrap();
+    update.with_plan(plan).unwrap();
+    let messages = update
+        .update_by_arrow_with_row_id(vec![update_batch(0)])
+        .await
+        .unwrap();
+    assert_eq!(messages.len(), 1);
+    assert_eq!(messages[0].check_from_snapshot, Some(2));
+    let deletes = update.delete_by_row_id(vec![1]).await.unwrap();
+    assert_eq!(deletes.len(), 1);
+    assert_eq!(deletes[0].check_from_snapshot, Some(2));
+    let error = update
+        .update_by_arrow_with_row_id(vec![update_batch(2)])
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("No file found for _ROW_ID 2"));
+}
+
+#[tokio::test]
+async fn planned_mutations_reject_foreign_paths_and_selected_splits_before_writing() {
+    let table = table("planned_mutation_validation").await;
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    let split = &plan.splits()[0];
+    for (path, snapshot_id, ranges, streaming) in [
+        ("memory:/another/bucket-0", 2, None, false),
+        (split.bucket_path(), 1, None, false),
+        (
+            split.bucket_path(),
+            2,
+            Some(vec![RowRange::new(0, 0)]),
+            false,
+        ),
+        (split.bucket_path(), 2, None, true),
+    ] {
+        let mut builder = DataSplitBuilder::new()
+            .with_snapshot(snapshot_id)
+            .with_partition(split.partition().clone())
+            .with_bucket(split.bucket())
+            .with_bucket_path(path.to_string())
+            .with_data_files(split.data_files().to_vec())
+            .with_streaming(streaming);
+        if let Some(ranges) = ranges {
+            builder = builder.with_row_ranges(ranges);
+        }
+        let split = builder.build().unwrap();
+        let mut update = table.new_write_builder().new_update().unwrap();
+        if update
+            .with_plan(Plan::from_splits(Some(2), vec![split]))
+            .is_err()
+        {
+            continue;
+        }
+        assert!(update.new_update_by_row_id().await.is_err());
+        assert!(update.delete_by_row_id(vec![0]).await.is_err());
+    }
+}
+
+#[tokio::test]
+async fn planned_update_rejects_captured_deleted_rows_but_allows_live_rows() {
+    let table = table("planned_update_deletion_vector").await;
+    let update = table.new_write_builder().new_update().unwrap();
+    let deletes = update.delete_by_row_id(vec![0]).await.unwrap();
+    table
+        .new_write_builder()
+        .new_commit()
+        .commit(deletes)
+        .await
+        .unwrap();
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    assert!(plan
+        .splits()
+        .iter()
+        .any(|split| split.data_deletion_files().is_some()));
+    let mut update = table.new_write_builder().new_update().unwrap();
+    update.with_plan(plan).unwrap();
+    let mut writer = update.new_update_by_row_id().await.unwrap();
+    let error = writer
+        .update_columns(vec![update_batch(0)], vec!["value".into()])
+        .await
+        .unwrap_err();
+    assert!(error
+        .to_string()
+        .contains("Cannot update deleted _ROW_ID 0"));
+    assert!(writer.commit_messages().is_empty());
+    let messages = writer
+        .update_columns(vec![update_batch(1)], vec!["value".into()])
+        .await
+        .unwrap();
+    assert_eq!(messages.len(), 1);
+    assert_eq!(messages[0].check_from_snapshot, Some(3));
+}
+
+#[tokio::test]
+async fn planned_delete_preserves_unselected_bucket_deletion_vectors() {
+    use futures::TryStreamExt;
+
+    let table = table("planned_delete_bucket_deletion_vectors").await;
+    let deletes = table
+        .new_write_builder()
+        .new_update()
+        .unwrap()
+        .delete_by_row_id(vec![0])
+        .await
+        .unwrap();
+    let previous_index = deletes[0].new_index_files[0].clone();
+    let previous_anchor = previous_index
+        .deletion_vectors_ranges
+        .as_ref()
+        .unwrap()
+        .keys()
+        .next()
+        .unwrap()
+        .clone();
+    table
+        .new_write_builder()
+        .new_commit()
+        .commit(deletes)
+        .await
+        .unwrap();
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    let routing = crate::table::RowIdPlan::from_plan(plan.clone()).unwrap();
+    // Route only the second anchor. The first anchor's existing deletion
+    // vector must survive replacement of the bucket's shared index file.
+    let selected = routing
+        .groups()
+        .iter()
+        .find(|(first, _, _)| *first == 1)
+        .unwrap()
+        .2
+        .clone();
+    let mut update = table.new_write_builder().new_update().unwrap();
+    update
+        .with_plan(Plan::from_splits(plan.snapshot_id(), vec![selected]))
+        .unwrap();
+    let messages = update.delete_by_row_id(vec![1]).await.unwrap();
+    assert_eq!(messages[0].check_from_snapshot, Some(3));
+    assert_eq!(messages[0].deleted_index_files, vec![previous_index]);
+    let ranges = messages[0].new_index_files[0]
+        .deletion_vectors_ranges
+        .as_ref()
+        .unwrap();
+    assert_eq!(ranges.len(), 2);
+    assert_eq!(ranges[&previous_anchor].cardinality, Some(1));
+    table
+        .new_write_builder()
+        .new_commit()
+        .commit(messages)
+        .await
+        .unwrap();
+    let builder = table.new_read_builder();
+    let plan = builder.new_scan().plan().await.unwrap();
+    let batches: Vec<RecordBatch> = builder
+        .new_read()
+        .unwrap()
+        .to_arrow(plan.splits())
+        .unwrap()
+        .try_collect()
+        .await
+        .unwrap();
+    assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 0);
+}
+
+#[tokio::test]
+async fn row_id_plan_routes_complete_groups_and_rejects_ambiguous_ranges() {
+    let table = table("planned_row_id_routing").await;
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    let routing = crate::table::RowIdPlan::from_plan(plan.clone()).unwrap();
+    assert_eq!(routing.snapshot_id(), Some(2));
+    assert_eq!(
+        routing
+            .groups()
+            .iter()
+            .map(|(first, last, _)| (*first, *last))
+            .collect::<Vec<_>>(),
+        vec![(0, 0), (1, 1)]
+    );
+    let selected = routing.groups()[0].2.clone();
+    let mut update = table.new_write_builder().new_update().unwrap();
+    update
+        .with_plan(Plan::from_splits(Some(2), vec![selected.clone()]))
+        .unwrap();
+    assert!(update
+        .delete_by_row_id(vec![1])
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("No file found for _ROW_ID 1"));
+    assert!(crate::table::RowIdPlan::from_plan(Plan::from_splits(
+        Some(2),
+        vec![selected.clone(), selected]
+    ))
+    .unwrap_err()
+    .to_string()
+    .contains("overlapping file groups"));
+    assert_eq!(routing.plan().splits(), plan.splits());
+}
+
+#[tokio::test]
+async fn row_id_plan_keeps_full_partial_columns_and_dedicated_subranges() {
+    let table = table("planned_row_id_partial_ranges").await;
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    let split = &plan.splits()[0];
+    let mut base = split.data_files()[0].clone();
+    base.first_row_id = Some(0);
+    base.row_count = 6;
+    base.max_sequence_number = 0;
+    let mut partial = base.clone();
+    partial.file_name = "partial-column.parquet".into();
+    partial.max_sequence_number = 1;
+    partial.write_cols = Some(vec!["value".into()]);
+    let mut left = partial.clone();
+    left.file_name = "left.blob".into();
+    left.row_count = 2;
+    left.max_sequence_number = 1;
+    left.write_cols = Some(vec!["value".into()]);
+    let mut right = left.clone();
+    right.file_name = "right.vector.store".into();
+    right.first_row_id = Some(2);
+    right.row_count = 4;
+    let files = vec![right, base.clone(), left.clone(), partial];
+    let split = split.for_data_evolution_files(files.clone());
+    let routing =
+        crate::table::RowIdPlan::from_plan(Plan::from_splits(Some(2), vec![split.clone()]))
+            .unwrap();
+    assert_eq!(routing.groups().len(), 1);
+    assert_eq!((routing.groups()[0].0, routing.groups()[0].1), (0, 5));
+    assert_eq!(routing.groups()[0].2.data_files().len(), 4);
+    assert!(files
+        .iter()
+        .all(|file| routing.groups()[0].2.data_files().contains(file)));
+    left.file_name = "invalid-short-column.parquet".into();
+    let invalid = split.for_data_evolution_files(vec![base, left]);
+    assert!(
+        crate::table::RowIdPlan::from_plan(Plan::from_splits(Some(2), vec![invalid]))
+            .unwrap_err()
+            .to_string()
+            .contains("identical ranges")
+    );
+}
+
+#[tokio::test]
+async fn captured_plan_rejects_other_operation_families_before_io_or_callbacks() {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    let table = table("planned_operation_family").await;
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .plan_all()
+        .await
+        .unwrap();
+    let mut update = table.new_write_builder().new_update().unwrap();
+    update.with_plan(plan).unwrap();
+    for directory in ["manifest", "snapshot"] {
+        table
+            .file_io()
+            .delete_dir(&format!("{}/{directory}", table.location()))
+            .await
+            .unwrap();
+    }
+    let calls = Arc::new(AtomicUsize::new(0));
+    let count = calls.clone();
+    let callback = crate::table::UpdateAssignment::Function(Arc::new(move |_| {
+        count.fetch_add(1, Ordering::SeqCst);
+        Ok(Vec::new())
+    }));
+    let errors = vec![
+        update
+            .new_shard_updater(0, 1, None)
+            .await
+            .err()
+            .unwrap()
+            .to_string(),
+        update
+            .update_by_predicate(None, vec![("value".into(), callback)], vec![])
+            .await
+            .unwrap_err()
+            .to_string(),
+        update
+            .upsert_by_arrow_with_key(vec![], vec!["value".into()])
+            .await
+            .unwrap_err()
+            .to_string(),
+        update
+            .merge_into(crate::table::MergeSource::SelfTable, vec![], vec![], vec![])
+            .await
+            .unwrap_err()
+            .to_string(),
+        update
+            .delete_by_predicate(None)
+            .await
+            .unwrap_err()
+            .to_string(),
+    ];
+    assert!(
+        errors
+            .iter()
+            .all(|error| error.contains("with_plan supports only row-ID")),
+        "{errors:?}"
+    );
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+}
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index bfc1ecf69..b85a2539b 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -26,6 +26,8 @@ use planning::plan_vindex_shards;
 use validation::{checked_i32, find_index_field, validate_vector_field};
 
 use crate::spec::{CoreOptions, Predicate};
+use crate::table::global_index_build_common::vector::VectorIndexShard;
+use crate::table::global_index_build_plan::GlobalIndexBuildPlan;
 use crate::table::{CommitMessage, RowRange, Table};
 use crate::vindex::{is_vindex_index_type, VindexVectorIndexOptions};
 use crate::{Error, Result};
@@ -60,13 +62,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         self
     }
 
-    pub(super) async fn prepare(
-        &self,
-    ) -> Result<(
-        Option<i64>,
-        Vec<CommitMessage>,
-        Vec<timing::VectorIndexBuildTiming>,
-    )> {
+    pub(super) async fn plan(&self) -> Result<GlobalIndexBuildPlan> {
         // Building the index scans the table's rows.
         CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
 
@@ -92,29 +88,30 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         let core_options = CoreOptions::new(&merged_options);
         super::global_index_build_common::validate_build_table(self.table)?;
         let rows_per_shard = core_options.global_index_row_count_per_shard()?;
-        let parallelism = core_options.global_index_build_parallelism()?;
+        core_options.global_index_build_parallelism()?;
 
         let index_field = find_index_field(self.table, index_column)?;
         validate_vector_field(index_field)?;
-        let vindex_options = VindexVectorIndexOptions::new(
+        let options = VindexVectorIndexOptions::new(
             self.table.schema().options(),
             &self.options,
             &self.index_type,
             index_field,
         )?;
-        let dimension = checked_i32(
-            vindex_options.dimension() as u64,
+        checked_i32(
+            options.dimension() as u64,
             "vindex dimension is too large for Rust builder",
         )?;
-        let index_meta =
-            serde_json::to_vec(&vindex_options.native_options).map_err(|e| Error::DataInvalid {
-                message: format!("Failed to serialize vindex options metadata: {e}"),
-                source: Some(Box::new(e)),
-            })?;
-
         let snapshot_manager = self.table.snapshot_manager();
         let Some(snapshot) = snapshot_manager.get_latest_snapshot().await? else {
-            return Ok((None, vec![], vec![]));
+            return Ok(GlobalIndexBuildPlan::new(
+                self.table,
+                None,
+                index_column.into(),
+                self.index_type.clone(),
+                self.options.clone(),
+                vec![],
+            ));
         };
 
         let mut read_builder = self.table.new_read_builder();
@@ -144,48 +141,93 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             rows_per_shard,
             &indexed,
         )?;
-        if shards.is_empty() {
-            return Ok((Some(snapshot.id()), vec![], vec![]));
+
+        if !shards.is_empty() {
+            crate::table::global_index_build_common::validate_existing_index_overlap(
+                self.table,
+                snapshot.index_manifest(),
+                &self.index_type,
+                index_field.id(),
+                None,
+                &shards
+                    .iter()
+                    .map(|shard| RowRange::new(shard.row_range_start, shard.row_range_end))
+                    .collect::<Vec<_>>(),
+            )
+            .await?;
         }
 
-        crate::table::global_index_build_common::validate_existing_index_overlap(
+        Ok(GlobalIndexBuildPlan::new(
             self.table,
-            snapshot.index_manifest(),
+            Some(snapshot),
+            index_column.into(),
+            self.index_type.clone(),
+            self.options.clone(),
+            shards,
+        ))
+    }
+
+    pub(super) async fn build_planned_shard(
+        &self,
+        shard: &VectorIndexShard,
+    ) -> Result<Option<(CommitMessage, Option<timing::VectorIndexBuildTiming>)>> {
+        let index_column = self
+            .index_column
+            .as_deref()
+            .ok_or_else(|| Error::DataInvalid {
+                message: "vindex index column is required".into(),
+                source: None,
+            })?;
+        let index_field = find_index_field(self.table, index_column)?;
+        validate_vector_field(index_field)?;
+        let options = VindexVectorIndexOptions::new(
+            self.table.schema().options(),
+            &self.options,
             &self.index_type,
-            index_field.id(),
-            None,
-            &shards
-                .iter()
-                .map(|shard| RowRange::new(shard.row_range_start, shard.row_range_end))
-                .collect::<Vec<_>>(),
-        )
-        .await?;
+            index_field,
+        )?;
+        let dimension = checked_i32(
+            options.dimension() as u64,
+            "vindex dimension is too large for Rust builder",
+        )?;
+        let index_meta =
+            serde_json::to_vec(&options.native_options).map_err(|error| Error::DataInvalid {
+                message: format!("Failed to serialize vindex options metadata: {error}"),
+                source: Some(Box::new(error)),
+            })?;
+        let built = self
+            .build_index_file(
+                shard,
+                index_column,
+                dimension,
+                index_field.id(),
+                &options,
+                index_meta,
+            )
+            .await?;
+        Ok(built.map(|built| {
+            let mut message = CommitMessage::new(shard.partition_bytes.clone(), 0, vec![]);
+            message.new_index_files = vec![built.meta];
+            (message, built.timing)
+        }))
+    }
 
+    pub(super) async fn prepare(
+        &self,
+    ) -> Result<(
+        Option<i64>,
+        Vec<CommitMessage>,
+        Vec<timing::VectorIndexBuildTiming>,
+    )> {
+        let plan = self.plan().await?;
+        let mut options = self.table.schema().options().clone();
+        options.extend(self.options.clone());
+        let parallelism = CoreOptions::new(&options).global_index_build_parallelism()?;
         let prepared = super::global_index_build_common::preparation::prepare_shards(
             self.table,
-            shards,
+            plan.shards(),
             parallelism,
-            |shard| {
-                let vindex_options = &vindex_options;
-                let index_meta = index_meta.clone();
-                async move {
-                    let built = self
-                        .build_index_file(
-                            &shard,
-                            index_column,
-                            dimension,
-                            index_field.id(),
-                            vindex_options,
-                            index_meta,
-                        )
-                        .await?;
-                    Ok(built.map(|built| {
-                        let mut message = CommitMessage::new(shard.partition_bytes, 0, vec![]);
-                        message.new_index_files = vec![built.meta];
-                        (message, built.timing)
-                    }))
-                }
-            },
+            |shard| async move { shard.build(self.table).await },
         )
         .await?;
         let mut messages = Vec::with_capacity(prepared.len());
@@ -196,8 +238,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 timings.push(timing);
             }
         }
-
-        Ok((Some(snapshot.id()), messages, timings))
+        Ok((plan.snapshot_id(), messages, timings))
     }
 }
 
diff --git a/crates/paimon/src/table/vindex_index_build_builder/tests.rs b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
index 617a8dbcd..cba9244de 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
@@ -481,6 +481,157 @@ async fn write_vectors(table: &Table, ids: Vec<i32>, vectors: Vec<Vec<f32>>) {
         .unwrap();
 }
 
+#[tokio::test]
+async fn planned_global_index_validates_dimension_before_empty_snapshot() {
+    let path = "memory:/planned_global_index_empty_dimension";
+    let table = vindex_e2e_table(path, "2");
+    setup_dirs(table.file_io(), path).await;
+    let mut builder = table.new_global_index_build_builder();
+    builder
+        .with_index_type(IVF_FLAT_IDENTIFIER)
+        .with_index_column("embedding")
+        .with_options(HashMap::from([(
+            "ivf-flat.dimension".into(),
+            (i32::MAX as u64 + 1).to_string(),
+        )]));
+    for error in [
+        builder.plan().await.unwrap_err(),
+        builder.build().await.unwrap_err(),
+    ] {
+        assert!(error.to_string().contains("dimension is too large"));
+    }
+}
+
+#[tokio::test]
+async fn planned_global_index_shards_keep_snapshot_schema_and_physical_coverage() {
+    let path = "memory:/planned_global_index_shards";
+    let mut options = vindex_e2e_options("2");
+    options.insert("ivf-flat.nlist".into(), "1".into());
+    options.insert("deletion-vectors.enabled".into(), "true".into());
+    let table = test_table_with_io(
+        FileIOBuilder::new("memory").build().unwrap(),
+        path,
+        vindex_schema_builder(options).build().unwrap(),
+    );
+    setup_dirs(table.file_io(), path).await;
+    write_vectors(&table, vec![1, 2, 3, 4], vec![vec![1., 0.]; 4]).await;
+    let write = table.new_write_builder();
+    write
+        .new_commit()
+        .commit(
+            write
+                .new_update()
+                .unwrap()
+                .delete_by_row_id(vec![1])
+                .await
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let mut builder = table.new_global_index_build_builder();
+    builder
+        .with_index_type(IVF_FLAT_IDENTIFIER)
+        .with_index_column("embedding");
+    let plan = builder.plan().await.unwrap();
+    let snapshot_id = plan.snapshot_id().unwrap();
+    let plan: crate::table::GlobalIndexBuildPlan =
+        serde_json::from_slice(&serde_json::to_vec(&plan).unwrap()).unwrap();
+    let shards = plan.shards();
+    assert_eq!(
+        shards.iter().map(|s| s.row_range()).collect::<Vec<_>>(),
+        vec![RowRange::new(0, 1), RowRange::new(2, 3)]
+    );
+    assert!(shards
+        .iter()
+        .all(|shard| shard.snapshot_id() == snapshot_id));
+    write_vectors(&table, vec![5, 6], vec![vec![0., 1.]; 2]).await;
+    // Changing the builder after planning cannot change a worker's source or field.
+    builder
+        .with_index_column("id")
+        .with_options(HashMap::from([("ivf-flat.dimension".into(), "999".into())]));
+    let messages = builder.build_shard(&shards[0]).await.unwrap();
+    assert_eq!(messages.len(), 1);
+    let file = &messages[0].new_index_files[0];
+    assert_eq!(
+        file.row_count, 2,
+        "deleted vectors remain in physical build coverage"
+    );
+    let meta = file.global_index_meta.as_ref().unwrap();
+    assert_eq!(
+        (
+            meta.row_range_start,
+            meta.row_range_end,
+            meta.index_field_id
+        ),
+        (0, 1, 1)
+    );
+    assert_eq!(
+        latest_vindex_index_files(&table).await.len(),
+        0,
+        "workers do not commit"
+    );
+    assert_eq!(
+        table
+            .file_io()
+            .list_status(&format!("{path}/index/"))
+            .await
+            .unwrap()
+            .iter()
+            .filter(|status| status.path.contains("vector-ivf-flat-global-index-"))
+            .count(),
+        1,
+        "one task must not build another shard or later append"
+    );
+    write.new_commit().commit(messages).await.unwrap();
+    let plan = table
+        .new_global_index_build_builder()
+        .with_index_type(IVF_FLAT_IDENTIFIER)
+        .with_index_column("embedding")
+        .plan()
+        .await
+        .unwrap();
+    assert_eq!(
+        plan.shards()
+            .iter()
+            .map(|s| s.row_range())
+            .collect::<Vec<_>>(),
+        vec![RowRange::new(2, 3), RowRange::new(4, 5)]
+    );
+}
+
+#[tokio::test]
+async fn planned_global_index_shards_recheck_worker_authorization_and_table_identity() {
+    let path = "memory:/planned_global_index_authorization";
+    let table = vindex_e2e_table(path, "2");
+    setup_dirs(table.file_io(), path).await;
+    write_vectors(&table, vec![1, 2], vec![vec![1., 0.]; 2]).await;
+    let plan = table
+        .new_global_index_build_builder()
+        .with_index_type(IVF_FLAT_IDENTIFIER)
+        .with_index_column("embedding")
+        .plan()
+        .await
+        .unwrap();
+    let shard = &plan.shards()[0];
+    let denied = table.copy_with_options(HashMap::from([(
+        "query-auth.enabled".into(),
+        "true".into(),
+    )]));
+    let error = denied
+        .new_global_index_build_builder()
+        .build_shard(shard)
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("query-auth"));
+    let other = vindex_e2e_table("memory:/different_table", "2");
+    let error = other
+        .new_global_index_build_builder()
+        .build_shard(shard)
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("different table"));
+}
+
 #[tokio::test]
 async fn vector_build_with_deletion_vectors_preserves_physical_coverage() {
     for granule in [false, true] {