[python] Use typed MERGE clauses and named execution factories

bindings/python/project-description.md

@@ -85,6 +85,49 @@ updates, including incremental row-ID updaters, expose the same chainable setter
 `with_commit_user` is available on batch, stream, and postpone fixed bucket write
 builders, and the identity applies to both writers and committers from that builder.
 
+### Typed MERGE Actions
+
+Batch and stream `merge_into` accept ordered `WhenMatched` and `WhenNotMatched`
+objects. Assignments use `MergeAssignment.source_column`, `target_column`,
+`literal`, or `function`. A function receives the selected clause's Arrow table
+and returns an Array or ChunkedArray with the same number of rows.
+
+Execution engines that already joined target and source data can reuse these
+types through the batch execution factories:
+
+```python
+import pyarrow as pa
+from pypaimon_rust.datafusion import MergeAssignment, MergeBatchExecution, WhenMatched
+
+joined = pa.table({
+    "t._ROW_ID": [10, 11],
+    "t.value": pa.array([20, 30], type=pa.int32()),
+    "s.value": pa.array([40, 10], type=pa.int32()),
+})
+matched = [
+    WhenMatched.update(
+        {"value": MergeAssignment.source_column("value")},
+        condition='"s.value" > "t.value"',
+    ),
+    WhenMatched.delete(),
+]
+update_schema = pa.schema([
+    pa.field("_ROW_ID", pa.int64(), nullable=False),
+    pa.field("value", pa.int32()),
+])
+updates = MergeBatchExecution.for_updates(joined.schema, update_schema, matched)
+deletes = MergeBatchExecution.for_deletes(joined.schema, matched)
+assert updates.execute(joined).to_pylist() == [{"_ROW_ID": 10, "value": 40}]
+assert deletes.execute(joined).to_pylist() == [{"_ROW_ID": 11}]
+```
+
+Both matched factories consume the complete ordered branch list. The delete
+factory skips UPDATE assignments and callbacks while preserving their conditions.
+For unmatched source batches, use `for_inserts(source_schema, target_schema,
+when_not_matched)` with `WhenNotMatched` objects. The factories compile conditions
+once and return an executor with `execute(Table | RecordBatch) -> Table`.
+The calling engine owns joins, authorization and table writes.
+
 ### Multimodal SQL Helpers
 
 `SQLContext` also registers Python scalar UDFs for common BLOB media and vector

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -18,7 +18,7 @@
 from datetime import datetime
 from os import PathLike
 from typing import (
-    Any, Callable, Dict, Iterable, List, Literal, Optional, Sequence, Tuple,
+    Any, Callable, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Tuple,
     TypeAlias, Union,
 )
 
@@ -79,6 +79,40 @@ class RowIdPlan:
         ...
     def plan(self) -> Plan: ...
 
+class MergeAssignment:
+    """An immutable source reference, target reference, literal or callback."""
+    @staticmethod
+    def source_column(name: str) -> "MergeAssignment": ...
+    @staticmethod
+    def target_column(name: str) -> "MergeAssignment": ...
+    @staticmethod
+    def literal(value: Any) -> "MergeAssignment":
+        """Defer scalar conversion; Arrow arrays provide values for selected rows."""
+        ...
+    @staticmethod
+    def function(
+        function: Callable[[pyarrow.Table], Union[pyarrow.Array, pyarrow.ChunkedArray]],
+    ) -> "MergeAssignment": ...
+
+class WhenMatched:
+    """An immutable, ordered UPDATE or DELETE clause with an optional SQL condition.
+
+    Conditions reference quoted Arrow names, such as ``"s.value" > "t.value"``.
+    The first matching clause consumes a row; NULL conditions fall through.
+    """
+    @staticmethod
+    def update(
+        assignments: Mapping[str, MergeAssignment], condition: Optional[str] = None,
+    ) -> "WhenMatched": ...
+    @staticmethod
+    def delete(condition: Optional[str] = None) -> "WhenMatched": ...
+
+class WhenNotMatched:
+    """An immutable, ordered INSERT clause."""
+    def __init__(
+        self, assignments: Mapping[str, MergeAssignment], condition: Optional[str] = None,
+    ) -> None: ...
+
 class MergeBatchExecution:
     """Evaluate ordered MERGE actions over already joined Arrow batches.
 
@@ -89,10 +123,22 @@ class MergeBatchExecution:
     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: ...
+    @staticmethod
+    def for_updates(
+        input_schema: pyarrow.Schema, update_schema: pyarrow.Schema,
+        when_matched: Sequence[WhenMatched],
+    ) -> "MergeBatchExecution": ...
+    @staticmethod
+    def for_deletes(
+        input_schema: pyarrow.Schema, when_matched: Sequence[WhenMatched],
+    ) -> "MergeBatchExecution":
+        """Keep all matched conditions; skip UPDATE assignment conversion and evaluation."""
+        ...
+    @staticmethod
+    def for_inserts(
+        input_schema: pyarrow.Schema, target_schema: pyarrow.Schema,
+        when_not_matched: Sequence[WhenNotMatched],
+    ) -> "MergeBatchExecution": ...
     def execute(self, batch: Union[pyarrow.Table, pyarrow.RecordBatch]) -> pyarrow.Table:
         """Return selected actions; unselected assignments are never evaluated."""
         ...
@@ -542,15 +588,14 @@ class BatchTableUpdate:
     ) -> "BatchTableUpdate": ...
     def merge_into(
         self, source: Optional[Union[pyarrow.Table, Table]], *, on: Sequence[Tuple[str, str]],
-        when_matched: Sequence[Dict[str, Any]], when_not_matched: Sequence[Dict[str, Any]]
+        when_matched: Sequence[WhenMatched], when_not_matched: Sequence[WhenNotMatched]
     ) -> List[CommitMessage]:
         """Prepare MERGE against one target snapshot. None selects self-merge on _ROW_ID.
 
         A Table source reads its selected snapshot or numeric timestamp window
         in core, independently of the target. Arrow input preserves batches and source row order.
 
-        Clauses carry assignments as (column, source/target/literal/callable, value)
-        tuples, a delete flag, and optional SQL condition metadata. Callables
+        Typed clauses carry MergeAssignment values and optional SQL conditions. Callables
         consume the complete clause's qualified t./s. Arrow columns and return
         an Array or ChunkedArray with the same row count; empty clauses are lazy.
         Commit returned messages through the matching builder's committer.
@@ -616,16 +661,15 @@ class StreamTableUpdate:
     ) -> "StreamTableUpdate": ...
     def merge_into(
         self, source: Optional[Union[pyarrow.Table, Table]], *, on: Sequence[Tuple[str, str]],
-        when_matched: Sequence[Dict[str, Any]], when_not_matched: Sequence[Dict[str, Any]],
+        when_matched: Sequence[WhenMatched], when_not_matched: Sequence[WhenNotMatched],
         commit_identifier: int
     ) -> List[CommitMessage]:
         """Prepare MERGE against one target snapshot. None selects self-merge on _ROW_ID.
 
         A Table source reads its selected snapshot or numeric timestamp window
         in core, independently of the target. Arrow input preserves batches and source row order.
 
-        Clauses carry assignments as (column, source/target/literal/callable, value)
-        tuples, a delete flag, and optional SQL condition metadata. Callables
+        Typed clauses carry MergeAssignment values and optional SQL conditions. Callables
         consume the complete clause's qualified t./s. Arrow columns and return
         an Array or ChunkedArray with the same row count; empty clauses are lazy.
         Commit returned messages through the matching builder's committer.

bindings/python/src/context.rs

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

bindings/python/src/lib.rs

@@ -31,6 +31,7 @@ mod hybrid_search;
 mod index_build;
 mod merge_execution;
 mod merge_into;
+mod merge_types;
 mod oss_cpp;
 mod predicate;
 mod read;