[python] Expose commit_mode and options on the public write_paimon

Actor mode was only reachable by importing write_paimon_dataset from
pypaimon.write.ray_datasink; write_paimon, the documented entry point,
had no way to select it or to pass dynamic table options.

Forward both kwargs and document what actor mode trades away: a partial
write stays committed if a later task fails, and the returned snapshot
id is the table's latest rather than this write's own.

Co-Authored-By: Claude Code <noreply@anthropic.com>

Author: wangzeyu03

paimon-python/pypaimon/ray/ray_paimon.py

@@ -545,6 +545,8 @@ def write_paimon(
     concurrency: Optional[int] = None,
     ray_remote_args: Optional[Dict[str, Any]] = None,
     hash_fixed_precluster: str = "auto",
+    commit_mode: str = "two_phase",
+    options: Optional[Dict[str, str]] = None,
 ) -> "Optional[PaimonWriteResult]":
     """Write a Ray Dataset to a Paimon table.
 
@@ -567,10 +569,26 @@ def write_paimon(
         hash_fixed_precluster: Pre-clustering mode. ``"auto"`` follows
             table options, ``"off"`` disables it, and ``"map_groups"``
             explicitly enables HASH_FIXED grouping.
+        commit_mode: ``"two_phase"`` (default) has the write tasks return
+            their commit messages to the driver, which commits once after
+            every task finishes, so the write is all-or-nothing and the
+            returned snapshot is exactly the one it committed.
+            ``"actor"`` instead has each task commit through a singleton
+            Ray actor, which serializes the commits and makes each task's
+            rows visible as soon as that task returns -- at the cost of a
+            partial write staying committed if a later task fails, and of
+            the returned snapshot id being the table's latest rather than
+            this write's own. Actor mode is append-only and rejects
+            ``overwrite``, ``hash_fixed_precluster="map_groups"`` and
+            postpone-bucket tables.
+        options: Optional dynamic table options applied for this write
+            only, e.g. ``{"target-file-size": "256mb"}``.
 
     Returns:
-        Metadata for the exact committed snapshot, or ``None`` when no
-        snapshot was committed.
+        Metadata for the committed snapshot, or ``None`` when no snapshot
+        was committed. With ``commit_mode="actor"`` this is the table's
+        latest snapshot after the write, which a concurrent writer may
+        have produced; see ``commit_mode`` above.
     """
     _require_ray_data()
 
@@ -587,4 +605,6 @@ def write_paimon(
         concurrency=concurrency,
         ray_remote_args=ray_remote_args,
         hash_fixed_precluster=hash_fixed_precluster,
+        commit_mode=commit_mode,
+        options=options,
     )

paimon-python/pypaimon/tests/ray_actor_commit_test.py

@@ -296,6 +296,64 @@ def test_two_phase_mode_applies_options(self):
         )
         self.assertEqual(len(self._read_all(table)), 3)
 
+    def test_public_write_paimon_reaches_actor_mode(self):
+        """``commit_mode`` and ``options`` must be reachable through the
+        documented public entry point, not only through the internal
+        ``write_paimon_dataset``. ``file.format`` doubles as proof that
+        both kwargs really traversed the wrapper rather than being
+        accepted and dropped.
+        """
+        from pypaimon.ray.ray_paimon import write_paimon
+
+        table = self._create_append_table('test_public_actor')
+
+        ds = ray.data.from_arrow(
+            pa.Table.from_pydict(
+                {'id': [1, 2, 3], 'value': ['a', 'b', 'c']}, schema=SCHEMA
+            )
+        )
+
+        result = write_paimon(
+            ds, 'default.test_public_actor', {'warehouse': self.warehouse},
+            commit_mode='actor',
+            options={'file.format': 'orc'},
+        )
+
+        self.assertEqual(result.row_count, 3)
+
+        table_dir = os.path.join(
+            self.warehouse, 'default.db', 'test_public_actor'
+        )
+        self.assertTrue(
+            glob.glob(os.path.join(table_dir, '**', '*.orc'), recursive=True),
+            f'no .orc data files under {table_dir}',
+        )
+        self.assertFalse(
+            glob.glob(os.path.join(table_dir, '**', '*.parquet'), recursive=True),
+            'file.format=orc did not survive the public wrapper',
+        )
+        self.assertEqual(len(self._read_all(table)), 3)
+
+    def test_public_write_paimon_rejects_unknown_commit_mode(self):
+        """The wrapper must forward ``commit_mode`` to the validation that
+        rejects it, rather than silently defaulting to two_phase.
+        """
+        from pypaimon.ray.ray_paimon import write_paimon
+
+        self._create_append_table('test_public_bad_mode')
+
+        ds = ray.data.from_arrow(
+            pa.Table.from_pydict({'id': [1], 'value': ['a']}, schema=SCHEMA)
+        )
+
+        with self.assertRaises(ValueError) as ctx:
+            write_paimon(
+                ds, 'default.test_public_bad_mode',
+                {'warehouse': self.warehouse},
+                commit_mode='nonsense',
+            )
+        self.assertIn('nonsense', str(ctx.exception))
+
     # ------------------------------------------------------------------
     # Commit dedup (actor replay / whole-task retry)
     # ------------------------------------------------------------------