[python][ray] Add time-based resumable commits to write_paimon

docs/docs/pypaimon/ray-data.md

@@ -277,6 +277,60 @@ overlapping buckets or sequence numbers for those modes.
   HASH_FIXED grouping. This option does not enable HASH_DYNAMIC or
   CROSS_PARTITION primary-key writes.
 
+### Incremental write
+
+An existing Ray Dataset can commit target groups periodically:
+
+```python
+write_paimon(
+    updates,
+    "database_name.target",
+    {"warehouse": "/path/to/warehouse"},
+    commit_mode="incremental",
+    update_cols=["feature"],
+    commit_interval_seconds=600,
+)
+```
+
+Committed groups survive failure, but retrying recomputes the Dataset lineage.
+A plain Dataset does not accept `operation_id`. Pin the starting snapshot when
+it reads from the target table itself.
+
+Use `PaimonOffsetSource` to resume source progress. Put the replayable
+transformation, including inference, in `transform`:
+
+```python
+from pypaimon.ray import PaimonOffsetSource, write_paimon
+
+def build_updates(window):
+    return window.map_batches(infer, batch_format="pyarrow")
+
+source = PaimonOffsetSource(
+    "database_name.source",
+    projection=["id", "payload"],
+    transform=build_updates,
+)
+write_paimon(
+    source,
+    "database_name.target",
+    {"warehouse": "/path/to/warehouse"},
+    commit_mode="incremental",
+    update_cols=["feature"],
+    operation_id="feature-backfill-2026-07",
+    commit_interval_seconds=600,
+)
+```
+
+Retry with the same `operation_id` skips completed source windows; only an
+unfinished window may run again.
+
+Use a fixed-bucket, partial-update target. Source rows need unique primary keys
+and all `update_cols`; missing keys are inserted. Source splits run in bounded
+windows, so the actual interval may be longer. Concurrent data changes or schema
+changes fail; compaction is allowed. After success,
+`delete_write_paimon_checkpoint` releases retained snapshots. An `operation_id`
+must not be reused after changing the source transformation.
+
 ### `TableWrite.write_ray()` (lower-level)
 
 If you have already constructed a `table_write` from a write builder, you can

paimon-python/pypaimon/ray/__init__.py

@@ -30,6 +30,8 @@
 )
 from pypaimon.ray.update_by_row_id import update_by_row_id
 from pypaimon.ray.read_by_row_id import read_by_row_id
+from pypaimon.ray.offset_source import PaimonOffsetSource
+from pypaimon.ray.incremental_write import delete_write_paimon_checkpoint
 
 __all__ = [
     "read_paimon",
@@ -40,6 +42,8 @@
     "merge_into",
     "update_by_row_id",
     "read_by_row_id",
+    "PaimonOffsetSource",
+    "delete_write_paimon_checkpoint",
     "WhenMatched",
     "WhenNotMatched",
     "source_col",