[core] Share MERGE action evaluation for distributed execution

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -79,6 +79,24 @@ class RowIdPlan:
         ...
     def plan(self) -> Plan: ...
 
+class MergeBatchExecution:
+    """Evaluate ordered MERGE actions over already joined Arrow batches.
+
+    Input fields use the ``t.<column>`` and ``s.<column>`` aliases. This
+    operator evaluates expressions and assignments without reading or writing
+    a table; callers retain responsibility for joins and authorization.
+    Matched input carries ``t._ROW_ID`` or ``_ROW_ID``, never both. Update
+    output contains ``_ROW_ID`` and the requested target columns; delete output
+    contains only ``_ROW_ID``. Insert output uses the target schema.
+    """
+    def __init__(
+        self, input_schema: pyarrow.Schema, output_schema: pyarrow.Schema,
+        clauses: Sequence[Dict[str, Any]], mode: Literal["update", "delete", "insert"],
+    ) -> None: ...
+    def execute(self, batch: Union[pyarrow.Table, pyarrow.RecordBatch]) -> pyarrow.Table:
+        """Return selected actions; unselected assignments are never evaluated."""
+        ...
+
 class ScanStats:
     """Read-only counters from normal metadata planning."""
     @property

bindings/python/src/context.rs

@@ -1038,6 +1038,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     this.add_class::<crate::write::PyBatchTableWrite>()?;
     this.add_class::<crate::write::PyBatchTableUpdate>()?;
     this.add_class::<crate::write::PyTableUpdateByRowId>()?;
+    this.add_class::<crate::merge_execution::PyMergeBatchExecution>()?;
     this.add_class::<crate::shard_update::PyShardTableUpdator>()?;
     this.add_class::<crate::write::PyStreamTableUpdate>()?;
     this.add_class::<crate::write::PyStreamTableWrite>()?;

bindings/python/src/lib.rs

@@ -29,6 +29,7 @@ mod full_text_search;
 mod global_index_result;
 mod hybrid_search;
 mod index_build;
+mod merge_execution;
 mod merge_into;
 mod oss_cpp;
 mod predicate;