[Python] Execute captured Ray mutation and index plans through native core

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

@@ -26,9 +26,27 @@ def __init__(self, native_commit, is_batch=False):
     def commit(self, messages, commit_identifier, snapshot_properties=None):
         kwargs = ({} if snapshot_properties is None else
                   {'snapshot_properties': snapshot_properties})
-        if self._is_batch:
-            return self._native_commit.commit(messages, **kwargs)
-        return self._native_commit.commit(commit_identifier, messages, **kwargs)
+        try:
+            if self._is_batch:
+                return self._native_commit.commit(messages, **kwargs)
+            return self._native_commit.commit(commit_identifier, messages, **kwargs)
+        except Exception as error:
+            self._raise_commit_error(error)
+
+    def check_row_id_conflicts(self, messages, latest_snapshot_id, check_from_snapshot):
+        try:
+            return self._native_commit.check_row_id_conflicts(messages, latest_snapshot_id, check_from_snapshot)
+        except Exception as error:
+            self._raise_commit_error(error)
+
+    @staticmethod
+    def _raise_commit_error(error):
+        from pypaimon.execution.native.write.native_commit import native_commit_error
+
+        converted = native_commit_error(error)
+        if converted is error:
+            raise error
+        raise converted from error
 
     def abort(self, messages):
         return self._native_commit.abort(messages)

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

@@ -202,6 +202,40 @@ def from_native_commit_messages(table, messages):
     return decoded
 
 
+def native_commit_error(error):
+    """Keep retry classification structured; never infer it from error text."""
+    from pypaimon_rust import RowIdExistenceConflict as NativeRowIdConflict
+    from pypaimon.write.row_id_conflict import RowIdExistenceConflict
+
+    if not isinstance(error, NativeRowIdConflict):
+        return error
+    return RowIdExistenceConflict._from_fields(
+        file_name=error.file_name, first_row_id=error.first_row_id,
+        row_count=error.row_count, bucket=error.bucket,
+        current_first_row_id=error.current_first_row_id,
+        current_row_count=error.current_row_count)
+
+
+def prepare_native_row_id_conflict_check(table, messages):
+    """Prepare a check without creating Python publication or conflict state."""
+    from pypaimon.execution.native.write.commit import NativeTableCommit
+
+    if not native_messages_supported(table, messages):
+        return None
+    native = create_native_commit(table, 'row-id-conflict-check', batch=True)
+    if native is None:
+        return None
+    try:
+        converted = to_native_commit_messages(table, messages)
+    except BaseException:
+        try:
+            native.close()
+        except BaseException:
+            pass
+        raise
+    return NativeTableCommit(native, is_batch=True), converted
+
+
 def native_commit_callback(table, callbacks):
     """Translate the native notification into the existing Python context."""
     from pypaimon.manifest.schema.manifest_entry import ManifestEntry

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

@@ -15,15 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
-"""Local global-index builds delegated to Rust, retaining Python's commit API."""
+"""Global-index builds delegated to Rust, retaining Python's commit API."""
 
 from pypaimon.execution.native.table import (_native_table, _option_value_to_string)
 from pypaimon.execution.native.read.builder import (_predicate_to_native)
 from pypaimon.execution.native.runtime import (native_runtime_available)
 from pypaimon.execution.native.write.native_commit import _rest_catalog_supported, from_native_commit_messages
 
 
-def build_native_global_index(builder, partition_filter):
+def build_native_global_index(builder, partition_filter, *, execution="local", concurrency=None, ray_remote_args=None):
     """Return unpublished messages, or None before building for an ineligible table.
 
     Rust owns snapshot selection, range planning and index file generation.
@@ -77,4 +77,16 @@ def build_native_global_index(builder, partition_filter):
     native.with_options(options)
     if partition_filter is not None:
         native.with_partition_filter(_predicate_to_native(partition_filter))
+    if execution == "ray":
+        from pypaimon.ray.vector_index_build import build_native_vector_index
+        # Planning may authorize and capture inputs; its errors must not select
+        # another engine or restart a build that may already have side effects.
+        plan = native.plan()
+        return build_native_vector_index(table, plan, concurrency, ray_remote_args)
     return from_native_commit_messages(table, native.build())
+
+
+def build_native_index_shard(table, shard):
+    """Execute captured metadata in Core; workers never replan or publish."""
+    native = _native_table(table).new_global_index_build_builder()
+    return from_native_commit_messages(table, native.build_shard(shard))

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

@@ -84,21 +84,25 @@ def create_native_update(table, commit_user, columns):
     return NativeBatchTableUpdate(table, writer)
 
 
-def create_native_update_by_row_id(table, commit_user, commit_identifier):
+def create_native_update_by_row_id(table, commit_user, commit_identifier, plan=None):
     """Create a core updater sharing one snapshot across incremental calls."""
     # Columns are selected per call; one core updater pins the snapshot and
     # tracks overlaps across Parquet and dedicated Blob updates.
     if not _native_update_columns_supported(table, None):
         return None
-    native_table = _native_row_id_table(table)
+    native_table = _native_row_id_table(
+        table._copy_with_snapshot(plan.table._read_snapshot) if plan is not None else table)
     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)
-                  .new_update().new_update_by_row_id())
+        update = native_table.new_batch_write_builder().with_commit_user(commit_user).new_update()
     else:
-        writer = (native_table.new_stream_write_builder().with_commit_user(commit_user)
-                  .new_update().new_update_by_row_id(commit_identifier))
+        update = native_table.new_stream_write_builder().with_commit_user(commit_user).new_update()
+    if plan is not None:
+        from pypaimon.execution.native.write.row_id_plan import native_plan
+        update.with_plan(native_plan(plan.table, plan.plan))
+    writer = (update.new_update_by_row_id() if commit_identifier == BATCH_COMMIT_IDENTIFIER
+              else update.new_update_by_row_id(commit_identifier))
     return NativeTableUpdateByRowId(
         table, commit_user, commit_identifier, writer)
 
@@ -438,22 +442,26 @@ def canonical_paths(node):
     return NativePredicateTableUpdate(table, writer, native_predicate)
 
 
-def create_native_delete(table, commit_user):
+def create_native_delete(table, commit_user, plan=None):
     """Select Rust's deletion-vector writer for supported batch deletes."""
     if not table.options.deletion_vectors_enabled(False):
         return None
-    native_table = _native_row_id_table(table)
+    native_table = _native_row_id_table(
+        table._copy_with_snapshot(plan.table._read_snapshot) if plan is not None else table)
     if native_table is None:
         return None
     writer = (native_table.new_batch_write_builder()
               .with_commit_user(commit_user)
               .new_update())
+    if plan is not None:
+        from pypaimon.execution.native.write.row_id_plan import native_plan
+        writer.with_plan(native_plan(plan.table, plan.plan))
     return NativeBatchTableUpdate(table, writer)
 
 
-def create_native_row_id_delete(table):
+def create_native_row_id_delete(table, plan=None):
     """Adapt the direct deletion API to the existing core table updater."""
-    operation = create_native_delete(table, table.new_batch_write_builder().commit_user)
+    operation = create_native_delete(table, table.new_batch_write_builder().commit_user, plan)
     return None if operation is None else NativeTableDeleteByRowId(operation)
 
 

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

@@ -0,0 +1,66 @@
+# 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.
+
+
+"""Typed split transport for Core's captured row-ID plans."""
+
+from pypaimon.execution.native.runtime import native_runtime_available
+
+
+def _with_bucket_path(table, split):
+    from copy import copy
+    from pypaimon.globalindex.indexed_split import IndexedSplit
+    from pypaimon.read.query_auth_split import QueryAuthSplit
+    from pypaimon.read.split import DataSplit
+
+    if isinstance(split, QueryAuthSplit):
+        inner = _with_bucket_path(table, split.split)
+        return split if inner is split.split else QueryAuthSplit(inner, split.auth_result)
+    if isinstance(split, IndexedSplit):
+        inner = _with_bucket_path(table, split.data_split())
+        return split if inner is split.data_split() else IndexedSplit(inner, split.row_ranges(), split.scores())
+    if not isinstance(split, DataSplit) or split.bucket_path:
+        return split
+    # Java keeps the table's canonical bucket path even when individual files
+    # live elsewhere. Serialization retains each exact external/file path.
+    captured = copy(split)
+    captured.bucket_path = table.path_factory().bucket_path(
+        tuple(split.partition.values), split.bucket, canonical_partition=True)
+    return captured
+
+
+def native_plan(table, plan):
+    from pypaimon_rust.datafusion import Plan, Split
+    from pypaimon.read.split_serializer import serialize_split_v1
+    return Plan(plan.snapshot_id, [Split.deserialize(serialize_split_v1(_with_bucket_path(table, split)))
+                                   for split in plan.splits()])
+
+
+def create_native_row_id_plan(table, plan):
+    if not (table.options.native_write_enabled() or table.options.native_read_enabled()
+            or table.options.native_plan_enabled()) or not native_runtime_available():
+        return None
+    from pypaimon_rust.datafusion import RowIdPlan
+    return RowIdPlan(native_plan(table, plan))
+
+
+def python_groups(table, plan):
+    from pypaimon.execution.native.table import _partition_fields
+    from pypaimon.read.split_serializer import deserialize_split_v1
+    return [(start, end, deserialize_split_v1(split.serialize(), _partition_fields(table),
+                                              table.trimmed_primary_keys_fields))
+            for start, end, split in plan.groups()]

paimon-python/pypaimon/execution/python/write/commit/conflict_detection.py

@@ -28,6 +28,7 @@
 from pypaimon.table.special_fields import SpecialFields
 from pypaimon.utils.range import Range
 from pypaimon.utils.range_helper import RangeHelper
+from pypaimon.write.row_id_conflict import RowIdExistenceConflict
 from pypaimon.execution.python.write.commit.commit_scanner import CommitScanner
 
 
@@ -165,23 +166,6 @@ def __init__(self, range_, field_ids):
         self.field_ids = field_ids
 
 
-class RowIdExistenceConflict(RuntimeError):
-    """A staged row-id file no longer matches the current base-file layout."""
-
-    def __init__(self, entry):
-        self.entry = entry
-        super().__init__(
-            "Row ID existence conflict: file '{}' references "
-            "firstRowId={}, rowCount={} in 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.".format(
-                entry.file.file_name,
-                entry.file.first_row_id,
-                entry.file.row_count,
-                entry.bucket))
-
-
 class ConflictDetection:
     """Detects conflicts between base and delta files during commit."""
 
@@ -489,6 +473,16 @@ def check_row_id_existence(self, base_entries, delta_entries, next_row_id=None):
         if not self.data_evolution_enabled:
             return None
 
+        # File identity excludes row IDs. A DELETE must still describe the
+        # current ADD before identifier-based manifest merging cancels it.
+        deleted = {entry.identifier(): entry for entry in delta_entries if entry.kind == 1}
+        for current in base_entries:
+            stale = deleted.get(current.identifier()) if current.kind == 0 else None
+            if stale is not None and (
+                    current.file.first_row_id != stale.file.first_row_id
+                    or current.file.row_count != stale.file.row_count):
+                return RowIdExistenceConflict._from_delete(stale, current)
+
         if next_row_id is None:
             return None
 

paimon-python/pypaimon/execution/python/write/commit/row_id_conflict.py

@@ -0,0 +1,55 @@
+# 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.
+
+"""Read-only logical conflict checks for the selected Python backend."""
+
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.execution.python.write.commit.commit_scanner import CommitScanner
+from pypaimon.execution.python.write.commit.conflict_detection import ConflictDetection
+from pypaimon.table.row.generic_row import GenericRow
+
+
+def check_row_id_conflicts(table, messages, latest_snapshot_id, check_from_snapshot):
+    if not table.options.data_evolution_enabled():
+        raise NotImplementedError('Row-ID conflict checks require a data-evolution table')
+    if any(message.compact_before or message.compact_after or message.compact_changelog_files
+           or message.compact_index_adds or message.compact_index_deletes for message in messages):
+        raise NotImplementedError('Row-ID conflict checks do not accept compact increments')
+    manager = table.snapshot_manager()
+    latest = manager.get_snapshot_by_id(latest_snapshot_id)
+    baseline = manager.get_snapshot_by_id(check_from_snapshot)
+    if latest is None or baseline is None:
+        raise ValueError('Row-ID conflict check snapshots must exist in the current table')
+    current_schema = table.schema_manager.latest()
+    if (baseline.schema_id != latest.schema_id
+            or current_schema is not None and current_schema.id != table.table_schema.id):
+        raise ValueError('Row-ID conflict checks require an unchanged, current table schema')
+    entries = []
+    for message in messages:
+        partition = GenericRow(list(message.partition), table.partition_keys_fields)
+        for kind, files in ((0, message.new_files), (1, message.deleted_files)):
+            entries.extend(ManifestEntry(
+                kind, partition, message.bucket,
+                table.options.bucket() if message.total_buckets is None else message.total_buckets, file)
+                for file in files)
+    manifests = ManifestListManager(table)
+    detection = ConflictDetection(True, manager, manifests, table, CommitScanner(table, manifests))
+    detection._row_id_check_from_snapshot = check_from_snapshot
+    conflict = detection.check_row_id_from_snapshot(latest, entries, check_compaction=False)
+    if conflict is not None:
+        raise conflict

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

@@ -0,0 +1,42 @@
+# 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.
+
+
+"""Python grouping for distributed row-ID plans, selected before execution."""
+
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.read.query_auth_split import QueryAuthSplit
+
+
+def _file_groups(split):
+    if isinstance(split, QueryAuthSplit):
+        for group in _file_groups(split.split):
+            yield QueryAuthSplit(group, split.auth_result)
+    else:
+        yield from PythonTableUpdate._predicate_update_file_groups([split])
+
+
+def row_id_groups(plan):
+    groups = []
+    for planned in plan.splits():
+        for split in _file_groups(planned):
+            ranges = [file.row_id_range() for file in split.files
+                      if file.first_row_id is not None and not DataFileMeta.is_blob_file(file.file_name)]
+            if ranges:
+                groups.append((min(r.from_ for r in ranges), max(r.to for r in ranges), split))
+    return sorted(groups, key=lambda group: group[0])

paimon-python/pypaimon/globalindex/index_dispatch.py

@@ -83,9 +83,11 @@ def close(self):
 
 
 def build_global_index(request, partition_filter, *, execution="local", concurrency=None, ray_remote_args=None):
-    if execution == "local" and request._table.options.native_write_enabled():
+    if request._table.options.native_write_enabled():
         from pypaimon.execution.native.write.native_index_build import build_native_global_index
-        messages = build_native_global_index(request, partition_filter)
+        messages = build_native_global_index(
+            request, partition_filter, execution=execution, concurrency=concurrency,
+            ray_remote_args=ray_remote_args)
         if messages is not None:
             return messages
     from pypaimon.execution.python.globalindex.create_global_index import PythonGlobalIndexBuilder