[Python] Dispatch row-id mutations and isolate write executors

docs/docs/pypaimon/writing.md

@@ -72,6 +72,22 @@ Use `new_postpone_fixed_bucket_write_builder()` to write real buckets. New
 partitions are buffered until `prepare_commit()`; existing ones are incremental.
 This builder currently supports `bucket-function.type=default` only.
 
+### Commit Identity
+
+Write builders generate a commit user by default. Use `with_commit_user()` before
+creating writers, updates or committers to share an application identity across
+distributed tasks or restore a stream application:
+
+```python
+write_builder = table.new_stream_write_builder().with_commit_user('application-1')
+table_write = write_builder.new_write()
+table_commit = write_builder.new_commit()
+```
+
+Writers and their committer must use the same identity. Use different identities
+for independent applications. Batch and postpone fixed-bucket builders expose
+the same method. Changing a builder affects only operations created afterward.
+
 ### Native Write and Commit
 
 With the `pypaimon-rust` extension installed, set `write.native.enabled=true`
@@ -111,6 +127,11 @@ does not prevent Native updates on tables without VECTOR fields. These operation
 continue to write Parquet/BLOB files; an active dedicated format must still be
 supported by the Native path.
 
+The same Native selection applies to direct construction of `TableWrite`,
+`BatchTableWrite`, `StreamTableWrite` and `PostponeFixedBucketBatchTableWrite`.
+Their public classes, constructor signatures and Arrow/Pandas batch overrides
+are preserved. Python execution is initialized only when that backend is selected.
+
 `write_ray()` uses the shared Python coordinator and lets workers select Native
 writers. For postpone fixed-bucket writes, overwrite plans use the new input
 unless an explicit bucket plan was supplied. A complete primary-key

paimon-python/pypaimon/execution/README.md

@@ -105,6 +105,24 @@ require unrestricted whole-table authorization because they expose raw file
 metadata. Internal writer metadata restoration uses the selected executor's
 private scanner and retains its write authorization boundary.
 
+Direct `TableUpdateByRowId`, `TableDeleteByRowId` and `ShardTableUpdator`
+construction also selects one execution backend before loading target files.
+Direct `TableWrite`, batch/stream writers and postpone fixed-bucket writers
+use the same selection boundary as builders while retaining their public class
+and subclass hooks. Writer resources and update executors are created in
+`write/backend_dispatch.py`; the public classes only delegate to the selected
+implementation. Python-only nested writers use explicit Python factories.
+An operation that selected Python keeps its nested update/delete writers in
+Python, including MERGE actions. Private schema, routing and rewrite algorithms
+call their executor's helpers; public method overrides and user callbacks remain
+public hooks. Broadcast file indexes retain their captured snapshot and are
+consumed by the explicitly selected Python workers without per-worker rescans.
+
+Postpone bucket metadata is writer state. Native obtains it from the Core write
+builder, preserving per-file layout validation and write authorization without
+adding SELECT requirements. Shared input statistics and bucket arithmetic stay
+outside either backend.
+
 Manifest-source and follow-up-scanner callbacks belong to Python execution
 internals. Public scanner configuration selects the existing snapshot, delta
 or changelog strategies without accepting custom scanner callbacks. Explicit

paimon-python/pypaimon/execution/native/write/native_commit.py

@@ -56,7 +56,7 @@ def create_native_commit(table, commit_user, overwrite_partition=None, *, batch=
     if native_table is None:
         return None
     if batch or overwrite_partition is not None:
-        builder = native_table.new_batch_write_builder()._with_commit_user(commit_user)
+        builder = native_table.new_batch_write_builder().with_commit_user(commit_user)
         if overwrite_partition is not None:
             builder = builder.with_overwrite(overwrite_partition)
         return builder.new_commit()

paimon-python/pypaimon/execution/native/write/native_merge_into.py

@@ -120,7 +120,7 @@ def encode(clause):
     encoded_not_matched = [encode(clause) for clause in not_matched]
     if any(clause is None for clause in encoded_matched + encoded_not_matched):
         return None
-    builder = (native_table.new_batch_write_builder()._with_commit_user(commit_user)
+    builder = (native_table.new_batch_write_builder().with_commit_user(commit_user)
                if commit_identifier == BATCH_COMMIT_IDENTIFIER else
                native_table.new_stream_write_builder().with_commit_user(commit_user))
     writer = builder.new_update()

paimon-python/pypaimon/execution/native/write/native_update.py

@@ -77,7 +77,7 @@ def create_native_update(table, commit_user, columns):
     if native_table is None:
         return None
     writer = (native_table.new_batch_write_builder()
-              ._with_commit_user(commit_user)
+              .with_commit_user(commit_user)
               .new_update())
     if columns is not None:
         writer.with_update_type(columns)
@@ -94,7 +94,7 @@ def create_native_update_by_row_id(table, commit_user, commit_identifier):
     if native_table is None:
         return None
     if commit_identifier == BATCH_COMMIT_IDENTIFIER:
-        writer = (native_table.new_batch_write_builder()._with_commit_user(commit_user)
+        writer = (native_table.new_batch_write_builder().with_commit_user(commit_user)
                   .new_update().new_update_by_row_id())
     else:
         writer = (native_table.new_stream_write_builder().with_commit_user(commit_user)
@@ -111,7 +111,7 @@ def create_native_shard_updator(table, commit_user, projection, columns, shard,
     if native_table is None:
         return None
     builder = (native_table.new_stream_write_builder().with_commit_user(commit_user) if stream else
-               native_table.new_batch_write_builder()._with_commit_user(commit_user))
+               native_table.new_batch_write_builder().with_commit_user(commit_user))
     update = builder.new_update()
     read_builder = table.new_read_builder()
     if projection is not None:
@@ -379,11 +379,11 @@ def create_native_upsert(table, commit_user, data, keys, columns):
                    for name in key_names):
             return None
     writer = (native_table.new_batch_write_builder()
-              ._with_commit_user(commit_user)
+              .with_commit_user(commit_user)
               .new_update()
               .with_update_type(columns))
     if blob_rows is not None:
-        writer._with_blob_uri_reader_factory(blob_rows)
+        writer.with_blob_uri_reader_factory(blob_rows)
     return NativeTableUpsert(table, writer, keys, data, blob_rows)
 
 
@@ -401,7 +401,7 @@ def create_native_predicate_update(table, commit_user, predicate, columns=None):
         # reading and callback execution belong to the core operation below.
         native_table.new_read_builder().with_filter(native_predicate)
     writer = (native_table.new_batch_write_builder()
-              ._with_commit_user(commit_user).new_update())
+              .with_commit_user(commit_user).new_update())
     return NativePredicateTableUpdate(table, writer, native_predicate)
 
 
@@ -434,7 +434,7 @@ def canonical_paths(node):
         # Resolve all fields/operators before selecting the mutating operation.
         native_table.new_read_builder().with_filter(native_predicate)
     writer = (native_table.new_batch_write_builder()
-              ._with_commit_user(commit_user).new_update())
+              .with_commit_user(commit_user).new_update())
     return NativePredicateTableUpdate(table, writer, native_predicate)
 
 
@@ -446,11 +446,26 @@ def create_native_delete(table, commit_user):
     if native_table is None:
         return None
     writer = (native_table.new_batch_write_builder()
-              ._with_commit_user(commit_user)
+              .with_commit_user(commit_user)
               .new_update())
     return NativeBatchTableUpdate(table, writer)
 
 
+def create_native_row_id_delete(table):
+    """Adapt the direct deletion API to the existing core table updater."""
+    operation = create_native_delete(table, table.new_batch_write_builder().commit_user)
+    return None if operation is None else NativeTableDeleteByRowId(operation)
+
+
+class NativeTableDeleteByRowId:
+
+    def __init__(self, operation):
+        self._operation = operation
+
+    def delete(self, row_ids):
+        return self._operation.delete_by_row_id(row_ids)
+
+
 def _raise_native_row_id_error(error):
     detail = str(error)
     if 'duplicate UPDATE operations' in detail:
@@ -533,11 +548,11 @@ def _write_row_columns(self, data, column_names, blob_object_columns):
                 arrow_field = pa.field(name, pa.list_(entries), nullable=arrow_field.nullable)
             data = data.append_column(arrow_field, pa.array(
                 [blobs.encode(value, field.type) for value in values], type=arrow_field.type))
-        self.writer._with_blob_uri_reader_factory(blobs)
+        self.writer.with_blob_uri_reader_factory(blobs)
         try:
             return self.update_columns(data, column_names)
         finally:
-            self.writer._with_blob_uri_reader_factory(None)
+            self.writer.with_blob_uri_reader_factory(None)
 
 
 class NativeTableUpsert:

paimon-python/pypaimon/execution/native/write/native_write.py

@@ -55,7 +55,7 @@ def create_native_write(table, commit_user, static_partition=None, stream=False,
     if native_table is None:
         return None
     if fixed_bucket:
-        builder = native_table.new_postpone_fixed_bucket_write_builder()._with_commit_user(commit_user)
+        builder = native_table.new_postpone_fixed_bucket_write_builder().with_commit_user(commit_user)
         if bucket_plan is not None:
             builder = builder.with_bucket_plan(bucket_plan.to_arrow(table))
         if static_partition is not None:
@@ -65,7 +65,7 @@ def create_native_write(table, commit_user, static_partition=None, stream=False,
     if stream:
         builder = native_table.new_stream_write_builder().with_commit_user(commit_user)
     else:
-        builder = native_table.new_batch_write_builder()._with_commit_user(commit_user)
+        builder = native_table.new_batch_write_builder().with_commit_user(commit_user)
         if static_partition is not None:
             builder = builder.with_overwrite(static_partition)
     if restore_snapshot_id is not None:

paimon-python/pypaimon/execution/native/write/postpone_bucket.py

@@ -17,8 +17,8 @@
 
 """Read existing bucket layouts through Rust before distributed input work."""
 
-from pypaimon.execution.native.read.plan import native_plan
 from pypaimon.execution.native.runtime import native_runtime_available
+from pypaimon.execution.native.write.native_commit import create_native_write_table
 from pypaimon.write.bucket_planning import PostponeBucketPlanning
 
 
@@ -33,18 +33,14 @@ class NativePostponeBucketPlanner(PostponeBucketPlanning):
 
     @staticmethod
     def _load_bucket_metadata(table):
-        # PartitionStat does not carry total_buckets. Complete file planning
-        # preserves that layout in the shared Split v1 metadata without ever
-        # creating a Python scanner, reader or data writer.
-        known = {}
-        for split in native_plan(table, all_files=True).splits():
-            if split.bucket < 0 or split.total_buckets is None or split.total_buckets <= 0:
-                continue
-            partition = tuple(split.partition.values)
-            previous = known.get(partition)
-            if previous is not None and previous != split.total_buckets:
-                raise RuntimeError(
-                    "Partition {} has different total buckets {} and {}"
-                    .format(partition, previous, split.total_buckets))
-            known[partition] = split.total_buckets
-        return known
+        # Writer restoration uses the catalog's write identity. Query plans
+        # would add SELECT rules and can combine different physical layouts.
+        native_table = create_native_write_table(table)
+        if native_table is None:
+            raise RuntimeError('Native bucket metadata planning is unavailable')
+        plan = native_table.new_postpone_fixed_bucket_write_builder().resolve_bucket_plan()
+        # A partition key may itself be called total_buckets. Keep the
+        # positional schema used by the Core plan and with_bucket_plan input.
+        columns = [column.to_pylist() for column in plan.columns]
+        return {tuple(column[row] for column in columns[:-1]): columns[-1][row]
+                for row in range(plan.num_rows)}

paimon-python/pypaimon/execution/python/write/merge_into.py

@@ -35,7 +35,7 @@
 from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
 from pypaimon.table.special_fields import SpecialFields
 from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.table_write import BatchTableWrite, StreamTableWrite
+from pypaimon.write.backend_dispatch import create_python_write
 from pypaimon.write.table_delete import TableDeleteByRowId
 from pypaimon.write.merge_inputs import (
     _PrepareCtx,
@@ -477,9 +477,9 @@ def _prepare_commit_messages(
         update_snapshot_table = _copy_at_snapshot(
             table, base_snapshot.id if base_snapshot is not None else None
         )
-        from pypaimon.write.write_builder import _new_update_by_row_id
+        from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
 
-        updater = _new_update_by_row_id(
+        updater = TableUpdateByRowId._from_python(
             update_snapshot_table, commit_user, commit_identifier)
         update_msgs = updater.update_columns(
             update_table, list(update_cols_union)
@@ -490,7 +490,7 @@ def _prepare_commit_messages(
         delete_snapshot_table = _copy_at_snapshot(
             table, base_snapshot.id if base_snapshot is not None else None
         )
-        deleter = TableDeleteByRowId(delete_snapshot_table)
+        deleter = TableDeleteByRowId._from_python(delete_snapshot_table)
         row_id_name = SpecialFields.ROW_ID.name
         delete_msgs = deleter.delete(
             delete_table.column(row_id_name).to_pylist()
@@ -513,9 +513,7 @@ def _prepare_commit_messages(
 
 
 def _new_table_write(table, commit_user: str, commit_identifier: int):
-    if commit_identifier == BATCH_COMMIT_IDENTIFIER:
-        return BatchTableWrite(table, commit_user)
-    return StreamTableWrite(table, commit_user)
+    return create_python_write(table, commit_user, stream=commit_identifier != BATCH_COMMIT_IDENTIFIER)
 
 
 def _normalize_source(source: Any, on_columns=()) -> pa.Table:

paimon-python/pypaimon/execution/python/write/postpone_batch_table_write.py

@@ -20,7 +20,7 @@
 
 from pypaimon.write.commit_message import CommitMessage
 from pypaimon.execution.python.write.file_store_write import PostponeFixedBucketFileStoreWrite
-from pypaimon.write.postpone_bucket import PostponeBucketPlanner
+from pypaimon.execution.python.write.postpone_bucket import PythonPostponeBucketPlanner as PostponeBucketPlanner
 from pypaimon.execution.python.write.table_write import PythonTableWrite
 from pypaimon.execution.python.write.row_key_extractor import PostponeFixedBucketRowKeyExtractor
 from pypaimon.write.row_utils import (
@@ -55,7 +55,7 @@ def initialize(
         )
         writer._plan_provided = bucket_plan is not None
         writer._pending_inputs = []
-        PythonTableWrite(writer).initialize(table, commit_user, static_partition)
+        super().initialize(table, commit_user, static_partition)
 
     def _create_file_store_write(self, commit_user):
         writer = self.writer
@@ -100,25 +100,25 @@ def write_arrow(self, table: pa.Table, bucket=None):
         writer = self.writer
         if bucket is not None:
             raise ValueError('Precomputed bucket writes require HASH_FIXED or HASH_DYNAMIC tables')
-        table = writer._prepare_arrow_data(table)
-        if not writer._buffer_input(table):
-            return PythonTableWrite(writer).write_arrow(table)
+        table = self._prepare_arrow_data(table)
+        if not self._buffer_input(table):
+            return super().write_arrow(table)
         writer._pending_inputs.extend(
             ("batch", batch) for batch in table.to_batches())
 
     def write_arrow_batch(self, data: pa.RecordBatch, bucket=None):
         writer = self.writer
         if bucket is not None:
             raise ValueError('Precomputed bucket writes require HASH_FIXED or HASH_DYNAMIC tables')
-        data = writer._prepare_arrow_data(data)
-        if not writer._buffer_input(data):
-            return PythonTableWrite(writer).write_arrow_batch(data)
+        data = self._prepare_arrow_data(data)
+        if not self._buffer_input(data):
+            return super().write_arrow_batch(data)
         writer._pending_inputs.append(("batch", data))
 
     def write_row(self, row):
         writer = self.writer
         if writer._plan_provided:
-            return PythonTableWrite(writer).write_row(row)
+            return super().write_row(row)
 
         values_by_name = row_to_named_values(
             row, writer.table.table_schema.fields)
@@ -136,7 +136,7 @@ def write_row(self, row):
         partition = tuple(
             values_by_name[key] for key in writer.table.partition_keys)
         if writer._bucket_plan.contains(partition):
-            return PythonTableWrite(writer).write_row(row)
+            return super().write_row(row)
 
         arrow_row = row_values_to_arrow_table(
             values_by_name, writer.table.table_schema.fields, column_names)
@@ -170,14 +170,13 @@ def _flush_pending_inputs(self):
         writer._pending_inputs = []
         for input_type, value in inputs:
             if input_type == "batch":
-                PythonTableWrite(writer).write_arrow_batch(value)
+                super().write_arrow_batch(value)
             else:
-                PythonTableWrite(writer).write_row(value[0])
+                super().write_row(value[0])
 
     def prepare_commit(self, commit_identifier) -> List[CommitMessage]:
-        writer = self.writer
-        writer._flush_pending_inputs()
-        return PythonTableWrite(writer).prepare_commit(commit_identifier)
+        self._flush_pending_inputs()
+        return super().prepare_commit(commit_identifier)
 
     def _distributed_write_options(self, *, overwrite=False) -> Dict[str, Any]:
         writer = self.writer
@@ -188,9 +187,9 @@ def _distributed_write_options(self, *, overwrite=False) -> Dict[str, Any]:
     def close(self):
         writer = self.writer
         writer._pending_inputs = []
-        PythonTableWrite(writer).close()
+        super().close()
 
     def abort(self):
         writer = self.writer
         writer._pending_inputs = []
-        PythonTableWrite(writer).abort()
+        super().abort()