[python] Add LeRobot Dataset v3 import

.github/workflows/paimon-python-checks.yml

@@ -139,6 +139,10 @@ jobs:
             fi
             python -m pip install 'h5py>=3,<4'
             python -c "import h5py; print('h5py', h5py.__version__)"
+            if [[ "${{ matrix.python-version }}" == "3.10" ]]; then
+              python -m pip install './paimon-python[lerobot]'
+              python -c "import datasets, lerobot; print('datasets', datasets.__version__, 'lerobot', lerobot.__version__)"
+            fi
 
             if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
               # Exercise the 0.4 API in one lane until its wheel is published.

docs/docs/pypaimon/multimodal-api.mdx

@@ -318,6 +318,59 @@ HDF5 core itself has no Ray dependency.
 provenance columns, maintain a source ledger, skip prior inputs, or detect
 source drift. Calling it again with the same input appends the rows again.
 
+## Load LeRobot Dataset v3
+
+`load_from_lerobot` converts a LeRobot Dataset v3 from a local directory,
+FileIO URI, or Hugging Face repository to a normal Paimon table once. It
+derives the schema from `meta/info.json`, keeps one row per frame, preserves
+Episode/frame indices and task text, and commits all batches in one Snapshot.
+
+```shell
+pip install 'pypaimon[lerobot]'
+```
+
+```python
+result = conn.load_from_lerobot(
+    "robot_data",
+    "/data/lerobot_dataset",
+)
+print(result.episode_count, result.row_count, result.snapshot_id)
+```
+
+An existing local directory is always used directly. FileIO-supported directory
+URIs such as `oss://bucket/lerobot_dataset` use credentials from the explicit
+`source_options` argument, never from the target Catalog. Another non-path
+string is passed to the official LeRobot API as a Hugging Face `repo_id`.
+
+```python
+result = conn.load_from_lerobot(
+    "robot_data",
+    "oss://source-bucket/lerobot_dataset",
+    source_options={
+        "fs.oss.endpoint": "oss-cn-hangzhou.aliyuncs.com",
+        "fs.oss.accessKeyId": "SOURCE_ACCESS_KEY_ID",
+        "fs.oss.accessKeySecret": "SOURCE_ACCESS_KEY_SECRET",
+    },
+)
+```
+
+URI metadata and Parquet files are read directly through Paimon FileIO. The
+LeRobot 0.4 video decoder requires a local path, so only the active MP4 chunk
+for each video feature is cached temporarily and removed after import. Local
+space is bounded by the active video chunks, not the complete dataset.
+
+If the target is absent, it is created from metadata. Existing targets use the
+same strict schema validation and append semantics as `load_from_hdf5`.
+One-dimensional numbers map to `VECTOR`; higher-rank values map to nested
+`ARRAY`; `image` and `video` map to `BLOB`. Images retain their compressed
+bytes. Videos are decoded once and stored as per-frame PNG BLOBs, so the shared
+MP4 is not repeated in every row.
+
+Use `feature_mapping={"observation.state": "state"}` for explicit renames. An
+optional `transform` receives each Arrow table and must preserve its row count
+and Episode/frame/index order. Version 2.x, `uint64`, language event
+structures, and depth-video semantics are not supported in this first version.
+
 ## Overwrite
 
 `overwrite` accepts the same input formats as `add` and replaces existing data

paimon-python/README.md

@@ -31,6 +31,32 @@ pip3 install dist/*.tar.gz
 
 The command will install the package and core dependencies to your local Python environment.
 
+# LeRobot Dataset v3 to multimodal tables
+
+Install the optional dependency, then import a local, FileIO URI, or Hugging
+Face v3 dataset once. The target table is created from `meta/info.json` when
+absent; later calls append.
+
+```commandline
+pip install 'pypaimon[lerobot]'
+```
+
+```python
+import pypaimon.multimodal as pmm
+
+connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
+result = connection.load_from_lerobot(
+    "robot_data",
+    "/data/lerobot_dataset",
+)
+print(result.row_count, result.snapshot_id)
+```
+
+Each LeRobot frame becomes one Paimon row. Numeric vectors remain vectors,
+images use their existing compressed bytes, and video frames are decoded once
+and stored as per-frame PNG BLOBs; the source MP4 is never copied into every
+row.
+
 # HDF5 to multimodal tables
 
 HDF5 loading requires Python 3.8 or newer. Install the optional dependency and

paimon-python/pypaimon/multimodal/__init__.py

@@ -29,6 +29,7 @@
     Hdf5File,
     Hdf5LoadResult,
 )
+from pypaimon.multimodal.lerobot import LeRobotLoadResult
 from pypaimon.multimodal.table import (
     MultimodalTable,
     TextRoute,
@@ -47,6 +48,7 @@
     "BlobStore",
     "Hdf5File",
     "Hdf5LoadResult",
+    "LeRobotLoadResult",
     "MultimodalConnection",
     "MultimodalTable",
     "NoSuchKey",

paimon-python/pypaimon/multimodal/arrow_utils.py

@@ -0,0 +1,139 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared Arrow schema validation for multimodal format importers."""
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def strict_arrow_table(
+        data,
+        target_schema,
+        source_path,
+        batch_index,
+        format_name):
+    if isinstance(data, pa.RecordBatch):
+        table = pa.Table.from_batches([data])
+    elif isinstance(data, pa.Table):
+        table = data
+    else:
+        raise ValueError(
+            "%s transform must return Arrow data or an iterable of Arrow data."
+            % format_name)
+
+    missing = [
+        name for name in target_schema.names if name not in table.column_names
+    ]
+    if missing:
+        raise ValueError(
+            "%s batch %d from %s is missing columns: %s"
+            % (format_name, batch_index, source_path, missing))
+    extra = [
+        name for name in table.column_names if name not in target_schema.names
+    ]
+    if extra:
+        raise ValueError(
+            "%s batch %d from %s has unexpected columns: %s"
+            % (format_name, batch_index, source_path, extra))
+    if table.column_names != target_schema.names:
+        raise ValueError(
+            "%s batch %d from %s has columns in the wrong order: %s; "
+            "expected %s."
+            % (format_name, batch_index, source_path, table.column_names,
+               target_schema.names))
+    try:
+        _validate_nested_nullability(table, target_schema)
+        if table.schema.equals(target_schema, check_metadata=False):
+            return table
+        casted = table.cast(target_schema, safe=True)
+        _validate_nested_nullability(casted, target_schema)
+        return casted
+    except (ValueError, TypeError, NotImplementedError) as error:
+        raise ValueError(
+            "%s batch %d from %s cannot be converted to the table schema: %s"
+            % (format_name, batch_index, source_path, error)) from error
+
+
+def _validate_nested_nullability(table, schema):
+    for field, column in zip(schema, table.columns):
+        for chunk in column.chunks:
+            _validate_array_nullability(chunk, field, field.name)
+
+
+def _validate_array_nullability(array, field, path):
+    if not field.nullable and array.null_count:
+        raise ValueError(
+            "non-nullable field %s contains %d null value(s)"
+            % (path, array.null_count))
+
+    target_type = field.type
+    source_type = array.type
+    if (pa.types.is_list(target_type)
+            or pa.types.is_large_list(target_type)
+            or pa.types.is_fixed_size_list(target_type)):
+        if not (pa.types.is_list(source_type)
+                or pa.types.is_large_list(source_type)
+                or pa.types.is_fixed_size_list(source_type)):
+            return
+        _validate_array_nullability(
+            pc.list_flatten(array),
+            target_type.value_field,
+            "%s.%s" % (path, target_type.value_field.name),
+        )
+        return
+
+    if pa.types.is_map(target_type):
+        if not pa.types.is_map(source_type):
+            return
+        start = array.offsets[0].as_py()
+        stop = array.offsets[-1].as_py()
+        length = stop - start
+        offsets = pc.subtract(
+            array.offsets,
+            pa.scalar(start, type=array.offsets.type),
+        )
+        entries = pa.StructArray.from_arrays(
+            [array.keys.slice(start, length),
+             array.items.slice(start, length)],
+            fields=[source_type.key_field, source_type.item_field],
+        )
+        logical_entries = pc.list_flatten(pa.ListArray.from_arrays(
+            offsets,
+            entries,
+            mask=pc.is_null(array),
+        ))
+        _validate_array_nullability(
+            logical_entries.field(0), target_type.key_field,
+            "%s.%s" % (path, target_type.key_field.name))
+        _validate_array_nullability(
+            logical_entries.field(1), target_type.item_field,
+            "%s.%s" % (path, target_type.item_field.name))
+        return
+
+    if pa.types.is_struct(target_type):
+        if not pa.types.is_struct(source_type):
+            return
+        parent_valid = pc.is_valid(array) if array.null_count else None
+        for index, child_field in enumerate(target_type):
+            child = array.field(index)
+            if parent_valid is not None:
+                child = pc.filter(child, parent_valid)
+            _validate_array_nullability(
+                child,
+                child_field,
+                "%s.%s" % (path, child_field.name),
+            )

paimon-python/pypaimon/multimodal/connection.py

@@ -117,6 +117,29 @@ def load_from_hdf5(
             source_options=source_options,
         )
 
+    def load_from_lerobot(
+            self,
+            table_name: str,
+            source,
+            *,
+            transform=None,
+            feature_mapping=None,
+            batch_size: int = 1024,
+            options=None,
+            source_options=None):
+        """Import one LeRobot Dataset v3 in a single append commit."""
+        from pypaimon.multimodal.lerobot import load_from_lerobot
+        return load_from_lerobot(
+            self,
+            table_name,
+            source,
+            transform=transform,
+            feature_mapping=feature_mapping,
+            batch_size=batch_size,
+            options=options,
+            source_options=source_options,
+        )
+
     def drop_table(self, name: str, ignore_if_not_exists: bool = False):
         self.catalog.drop_table(
             self._identifier(name),