[core] Add captured mutation and index build plans with structured row-id conflicts

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()."""
         ...

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>()?;

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| {