[python] Dispatch captured row ID and bucket execution

paimon-python/dev/test_packaging.py

@@ -211,7 +211,7 @@ def test_distribution_contents(self):
             "pypaimon/globalindex/vindex_reader.py",
             "pypaimon/write/bucket_planning.py",
             "pypaimon/write/commit_options.py",
-            "pypaimon/write/input_routing.py",
+            "pypaimon/execution/python/write/input_routing.py",
         )
         for name in execution_files:
             self.assertIn(name, names)

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

@@ -64,3 +64,38 @@ def python_groups(table, plan):
     return [(start, end, deserialize_split_v1(split.serialize(), _partition_fields(table),
                                               table.trimmed_primary_keys_fields))
             for start, end, split in plan.groups()]
+
+
+class NativeRowIdPlan:
+    """Transport Arrow row IDs to Core without reading or replanning files."""
+
+    def __init__(self, table, plan, native=None, groups=None):
+        self.table = table
+        self.plan = plan
+        self._native = native
+        self.groups = groups if groups is not None else python_groups(table, self._ensure_native())
+
+    def _ensure_native(self):
+        if self._native is None:
+            from pypaimon_rust.datafusion import RowIdPlan
+            self._native = RowIdPlan(native_plan(self.table, self.plan))
+        return self._native
+
+    def route(self, row_ids):
+        return self._ensure_native().route(row_ids)
+
+    def route_by_bucket(self, row_ids):
+        return self._ensure_native().route_by_bucket(row_ids)
+
+    def select(self, row_ids):
+        from pypaimon.execution.native.read.plan import _from_native_plan
+        selected = _from_native_plan(self.table, self._ensure_native().select(row_ids))
+        return self.plan._with_splits(selected.splits())
+
+    def for_plan(self, plan, groups):
+        return NativeRowIdPlan(self.table, plan, groups=groups)
+
+    def __getstate__(self):
+        state = self.__dict__.copy()
+        state['_native'] = None
+        return state

paimon-python/pypaimon/execution/python/read/scanner/bucket_select_converter.py

@@ -30,7 +30,7 @@
 happen — that would be silent data loss.
 
 The hashing routine reuses ``_hash_bytes_by_words`` /
-``_bucket_from_hash`` from ``pypaimon.write.input_routing`` — the same
+``_bucket_from_hash`` from ``pypaimon.execution.python.write.input_routing`` — the same
 code path the writer uses to assign rows to buckets. Reusing it (rather
 than copying) is what guarantees read/write hash agreement in the face of
 future routine changes.
@@ -70,7 +70,7 @@
 from pypaimon.schema.data_types import DataField
 from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
 from pypaimon.table.row.internal_row import InternalRow, RowKind
-from pypaimon.write.input_routing import (
+from pypaimon.execution.python.write.input_routing import (
     _bucket_from_hash,
     _hash_bytes_by_words,
 )