[python] Import LeRobot video datasets

docs/docs/pypaimon/multimodal-api.mdx

@@ -556,11 +556,15 @@ snapshot_id = conn.load_from_lerobot(
 
 Missing tables are created from metadata; existing tables use strict schema
 validation and append semantics. Scalars map to scalar types, vectors to
-`VECTOR`, higher-rank tensors to nested `ARRAY`, and images to `BLOB`. Images
-keep their compressed bytes.
-
-Only v3 is supported. Video features, `uint64`, and language event structures
-are rejected.
+`VECTOR`, higher-rank tensors to nested `ARRAY`, and images or video frames to
+`BLOB`. Images keep their compressed bytes; videos pack encoded MP4 payloads
+instead of repeating them per frame, with frame ordinals in the video field.
+Video imports keep each Episode in one aligned file group. When the normal
+file or a camera sidecar needs to roll, all active writers close together at
+the next Episode boundary. The target table must be unpartitioned and
+bucket-unaware.
+
+Only v3 is supported. `uint64` and language event structures are rejected.
 
 ## Overwrite
 

paimon-python/pypaimon/multimodal/lerobot/api.py

@@ -34,6 +34,7 @@
     _require_v3,
     _schema_from_info,
     _validate_lerobot_schema,
+    _video_feature_names,
 )
 from pypaimon.multimodal.lerobot.source import (
     _has_tasks,
@@ -48,6 +49,13 @@
     _validate_source_kerberos,
 )
 from pypaimon.multimodal.table import _target_schema
+from pypaimon.table.bucket_mode import BucketMode
+
+
+_VIDEO_LAYOUT_ERROR = (
+    "LeRobot video import requires an unpartitioned, bucket-unaware "
+    "target table so each Episode is written by one writer."
+)
 
 
 def load_from_lerobot(
@@ -83,6 +91,7 @@ def load_from_lerobot(
         _require_v3(local_info, resolved_source.path)
         _validate_info_paths(local_info)
         _schema_from_info(local_info, include_task=False)
+        video_fields = _video_feature_names(local_info)
         total_frames, _, total_tasks = \
             _validated_counts(local_info, resolved_source.path)
         if total_frames == 0:
@@ -96,11 +105,16 @@ def load_from_lerobot(
                 source_schema,
                 options,
                 resolved_source,
+                video_fields,
             )
             return None
         LeRobotDataset = _import_lerobot_dataset()
         dataset = _open_resolved_dataset(
-            LeRobotDataset, resolved_source, local_info)
+            LeRobotDataset,
+            resolved_source,
+            local_info,
+            download_videos=bool(video_fields),
+        )
         try:
             info = dict(dataset.meta.info)
             _require_v3(info, resolved_source.path)
@@ -109,12 +123,14 @@ def load_from_lerobot(
 
             source_schema = _schema_from_info(
                 info, include_task=_has_tasks(dataset, info))
+            video_fields = _video_feature_names(info)
             table = _validated_table(
                 connection,
                 table_name,
                 source_schema,
                 options,
                 resolved_source,
+                video_fields,
             )
 
             if row_count == 0:
@@ -126,6 +142,7 @@ def load_from_lerobot(
                 resolved_source,
                 source_schema,
                 batch_size,
+                video_fields,
             )
         finally:
             close = getattr(dataset, "close", None)
@@ -160,9 +177,10 @@ def _required_count(info, name, source):
 
 
 def _validated_table(
-        connection, table_name, source_schema, options, source):
+        connection, table_name, source_schema, options, source,
+        video_fields=()):
     table = _get_or_create_table(
-        connection, table_name, source_schema, options)
+        connection, table_name, source_schema, options, video_fields)
     target_schema = _target_schema(table.raw_table)
     _validate_lerobot_schema(
         source_schema, target_schema, source.path)
@@ -172,13 +190,42 @@ def _validated_table(
         source,
         0,
     )
+    configured = table.raw_table.options.video_frame_fields()
+    if configured != set(video_fields):
+        raise ValueError(
+            "LeRobot video features %s require table option "
+            "'video-frame-field'=%r; found %s."
+            % (list(video_fields), ",".join(video_fields), sorted(configured))
+        )
+    if video_fields and (
+            table.raw_table.partition_keys
+            or table.raw_table.bucket_mode() != BucketMode.BUCKET_UNAWARE):
+        raise ValueError(_VIDEO_LAYOUT_ERROR)
     return table
 
 
-def _get_or_create_table(connection, table_name, schema, options):
+def _get_or_create_table(
+        connection, table_name, schema, options, video_fields=()):
     try:
         return connection.get_table(table_name)
     except (DatabaseNotExistException, TableNotExistException):
+        options = dict(options or {})
+        if video_fields and str(options.get("bucket", "-1")).strip() != "-1":
+            raise ValueError(_VIDEO_LAYOUT_ERROR)
+        configured = options.get("video-frame-field")
+        if configured is not None:
+            requested = {
+                name.strip() for name in str(configured).split(",")
+                if name.strip()
+            }
+            if requested != set(video_fields):
+                raise ValueError(
+                    "LeRobot video features %s do not match "
+                    "'video-frame-field'=%r."
+                    % (list(video_fields), configured)
+                )
+        if video_fields:
+            options["video-frame-field"] = ",".join(video_fields)
         return connection.create_table(
             table_name,
             schema=schema,