[python] Add as-of joins for multimodal scans

docs/docs/pypaimon/multimodal-api.mdx

@@ -811,6 +811,45 @@ with docs.scan().where("category = 'lake'").to_arrow_batch_reader() as reader:
         consume(batch)
 ```
 
+### As-of joins
+
+`join_asof` preserves each left row and matches at most one right row within the
+same `by` group. Chain joins to align multiple streams without materializing
+intermediate results.
+
+```python
+from datetime import timedelta
+from pypaimon.multimodal import join_asof
+
+steps = join_asof(
+    actions.scan().select(["episode_id", "event_time", "action"]),
+    images.scan().where("camera = 'left'").select("image"),
+    on="event_time",
+    by="episode_id",
+    direction="nearest",
+    tolerance=timedelta(milliseconds=20),
+).join_asof(
+    topics.scan().where("topic = '/robot/state'").select("value"),
+    direction="backward",
+    tolerance=timedelta(milliseconds=50),
+)
+
+for batch in steps.to_arrow_batch_reader(batch_size=128):
+    train(batch)
+```
+
+Alignment reads each right scan's `by` and temporal columns internally, even
+when the query selects only payload columns. Unique right column names are kept;
+conflicts use `suffix="_right"`. Unmatched right payloads are null. Use
+`right_on` when the right timestamp column has a different name.
+Temporal and grouping columns must be non-null and have matching types across
+scans; grouping columns must be scalar. `direction` supports `backward`,
+`forward`, and `nearest`; tolerance is optional and inclusive. A zero-tolerance
+nearest join is an exact match. Nearest ties choose the earlier row.
+
+Inputs are pinned to their current snapshots when `join_asof` is created, and
+BLOB values remain descriptors until explicitly read or decoded.
+
 ### Reading BLOB columns
 
 `scan().read_blobs(column)` bulk-fetches a BLOB column's bytes for the filtered

paimon-python/pypaimon/multimodal/__init__.py

@@ -41,6 +41,9 @@
     text_route,
     vector_route,
 )
+from pypaimon.multimodal.temporal import (
+    join_asof,
+)
 from pypaimon.multimodal.video import VideoFrameCollator
 from pypaimon.table.row.blob import Blob, BlobDescriptor, VideoFrameDescriptor
 from pypaimon.table.data_evolution_merge_into import (
@@ -69,6 +72,7 @@
     "VideoFrameCollator",
     "VideoFrameDescriptor",
     "connect",
+    "join_asof",
     "lit",
     "source_col",
     "target_col",