[python] Persist LeRobot dataset metadata in Paimon

docs/docs/pypaimon/multimodal-api.mdx

@@ -617,26 +617,31 @@ orphan files for normal Paimon cleanup.
 
 `load_from_lerobot` imports a local directory, FileIO URI, or Hugging Face
 repository. It derives the schema from `meta/info.json`, writes one row per
-frame, and commits once.
+frame, and creates a LeRobot dataset backed by the frame table,
+`<table>__versions`, `<table>__episodes`, `<table>__tasks`, and an optional
+`<table>__subtasks`. Task text remains in the metadata table; frames retain
+`task_index`. After all components are committed and tagged with the same
+numeric `version_id`, one row is appended to `<table>__versions` to publish the
+version.
 
 ```shell
 pip install 'pypaimon[lerobot]'
 ```
 
 ```python
-snapshot_id = conn.load_from_lerobot(
+version_id = conn.load_from_lerobot(
     "robot_data",
     "/data/lerobot_dataset",
 )
-print(snapshot_id)
+print(version_id)
 ```
 
-The return value is `None` when the source has no frames.
+The returned `version_id` is the common tag name for all dataset components.
 
 For FileIO URIs, pass credentials through `source_options`:
 
 ```python
-snapshot_id = conn.load_from_lerobot(
+version_id = conn.load_from_lerobot(
     "robot_data",
     "oss://source-bucket/lerobot_dataset",
     source_options={
@@ -647,13 +652,14 @@ 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.
+A row in `<table>__versions` identifies a published release. Readers must first
+resolve that row, then read every required component through its matching tag;
+a missing tag is an error and must not fall back to the latest snapshot.
+The component tags are immutable and must be retained or deleted together.
+The one-time importer requires a new target table.
 
-Only v3 is supported. Video features, `uint64`, and language event structures
-are rejected.
+Scalars map to scalar types, vectors to `VECTOR`, higher-rank tensors to nested
+`ARRAY`, and images to `BLOB`. Images keep their compressed bytes.
 
 ## Overwrite
 

paimon-python/README.md

@@ -44,15 +44,17 @@ pip install 'pypaimon[lerobot]'
 import pypaimon.multimodal as pmm
 
 connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
-snapshot_id = connection.load_from_lerobot(
+version_id = connection.load_from_lerobot(
     "robot_data",
     "/data/lerobot_dataset",
 )
-print(snapshot_id)
+print(version_id)
 ```
 
-The schema comes from `meta/info.json`. Each frame becomes one row; media uses
-BLOB columns. Missing tables are created and later calls append.
+The source dataset must be non-empty. Its schema comes from `meta/info.json`.
+Each frame becomes one row; media uses BLOB columns. The import creates frame,
+Episode, task, and version tables and tags the three component tables with the
+returned `version_id`.
 
 # HDF5 to multimodal tables
 

paimon-python/pypaimon/multimodal/connection.py

@@ -125,7 +125,7 @@ def load_from_lerobot(
             batch_size: int = 1024,
             options=None,
             source_options=None):
-        """Import LeRobot Dataset v3 and return the committed snapshot ID."""
+        """Import LeRobot Dataset v3 into a new Paimon table group."""
         from pypaimon.multimodal.lerobot import load_from_lerobot
         return load_from_lerobot(
             self,

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

@@ -20,23 +20,24 @@
 import sys
 from typing import Mapping, Optional
 
-import pyarrow as pa
-
-from pypaimon.catalog.catalog_exception import (
-    DatabaseNotExistException,
-    TableNotExistException,
-)
-from pypaimon.multimodal.lerobot.loader import (
-    _strict_lerobot_table,
-    _write_dataset,
+from pypaimon.catalog.catalog_exception import TableAlreadyExistException
+from pypaimon.multimodal.lerobot.metadata import (
+    _append_arrow_tables,
+    _load_dataset_metadata,
+    _managed_table_options,
+    _prepare_metadata_tables,
+    _positive_integer,
+    _publish_dataset,
+    _validated_episode_tables,
 )
+from pypaimon.multimodal.lerobot.loader import _write_dataset
 from pypaimon.multimodal.lerobot.schema import (
     _require_v3,
     _schema_from_info,
-    _validate_lerobot_schema,
+    _validate_v3_required_features,
 )
 from pypaimon.multimodal.lerobot.source import (
-    _has_tasks,
+    _close_quietly,
     _import_lerobot_dataset,
     _load_hub_info,
     _open_resolved_dataset,
@@ -47,7 +48,6 @@
     _validated_source_options,
     _validate_source_kerberos,
 )
-from pypaimon.multimodal.table import _target_schema
 
 
 def load_from_lerobot(
@@ -57,13 +57,14 @@ def load_from_lerobot(
         *,
         batch_size: int = 1024,
         options: Optional[Mapping[str, object]] = None,
-        source_options: Optional[Mapping[str, object]] = None):
-    """Import LeRobot Dataset v3 and return the committed snapshot ID.
-
-    A missing target table is created from LeRobot metadata. An existing table
-    receives the same strict schema validation and append semantics as
-    :meth:`MultimodalConnection.load_from_hdf5`. FileIO URI credentials come
-    only from ``source_options`` and are not inherited from the target Catalog.
+        source_options: Optional[Mapping[str, object]] = None,
+) -> int:
+    """Import LeRobot Dataset v3 and return its version ID.
+
+    A new target table is created from LeRobot metadata. Episode, task, and
+    version metadata are stored in companion Paimon tables.
+    FileIO URI credentials come only from ``source_options`` and are not
+    inherited from the target Catalog.
     """
     if sys.version_info < (3, 10):
         raise RuntimeError(
@@ -82,55 +83,79 @@ def load_from_lerobot(
             local_info = _load_hub_info(resolved_source)
         _require_v3(local_info, resolved_source.path)
         _validate_info_paths(local_info)
-        _schema_from_info(local_info, include_task=False)
-        total_frames, _, total_tasks = \
-            _validated_counts(local_info, resolved_source.path)
-        if total_frames == 0:
-            source_schema = _schema_from_info(
-                local_info,
-                include_task=total_tasks > 0,
-            )
-            _validated_table(
-                connection,
-                table_name,
-                source_schema,
-                options,
-                resolved_source,
-            )
-            return None
+        _schema_from_info(local_info)
+        _positive_integer(local_info.get("fps"), "fps")
+        _validated_counts(local_info, resolved_source.path)
+        _validate_v3_required_features(local_info)
         LeRobotDataset = _import_lerobot_dataset()
         dataset = _open_resolved_dataset(
             LeRobotDataset, resolved_source, local_info)
         try:
             info = dict(dataset.meta.info)
             _require_v3(info, resolved_source.path)
-            row_count, _, _ = \
-                _validated_counts(info, resolved_source.path)
+            _validated_counts(info, resolved_source.path)
+            _validate_v3_required_features(info)
 
-            source_schema = _schema_from_info(
-                info, include_task=_has_tasks(dataset, info))
-            table = _validated_table(
+            lerobot_schema = _schema_from_info(info)
+            metadata = _load_dataset_metadata(
+                dataset, info, resolved_source)
+            return _import_dataset(
                 connection,
                 table_name,
-                source_schema,
-                options,
-                resolved_source,
-            )
-
-            if row_count == 0:
-                return None
-            return _write_dataset(
-                table,
                 dataset,
                 info,
                 resolved_source,
-                source_schema,
+                lerobot_schema,
                 batch_size,
+                options,
+                metadata,
             )
         finally:
             close = getattr(dataset, "close", None)
             if callable(close):
-                close()
+                _close_quietly(dataset, "dataset")
+
+
+def _import_dataset(
+        connection,
+        table_name,
+        dataset,
+        info,
+        source,
+        source_schema,
+        batch_size,
+        options,
+        metadata):
+    table = _create_target_table(
+        connection, table_name, source_schema, options)
+    tables = _prepare_metadata_tables(
+        connection, table.raw_table, metadata)
+    version_id = 1
+    episodes_snapshot_id = _append_arrow_tables(
+        tables["episodes"],
+        _validated_episode_tables(metadata),
+    )
+    frames_snapshot_id = None
+    if int(info["total_frames"]) > 0:
+        frames_snapshot_id = _write_dataset(
+            table,
+            dataset,
+            info,
+            source,
+            source_schema,
+            batch_size,
+            metadata,
+        )
+    _publish_dataset(
+        connection,
+        tables,
+        version_id,
+        metadata,
+        table.identifier,
+        frames_snapshot_id,
+        episodes_snapshot_id,
+    )
+    return version_id
 
 
 def _validated_counts(info, source):
@@ -142,6 +167,9 @@ def _validated_counts(info, source):
             "LeRobot metadata %s has inconsistent counts: total_frames=%d "
             "and total_episodes=%d must both be zero or both be positive."
             % (source, total_frames, total_episodes))
+    if total_frames == 0:
+        raise ValueError(
+            "load_from_lerobot requires a non-empty LeRobot Dataset v3.")
     return total_frames, total_episodes, total_tasks
 
 
@@ -159,28 +187,26 @@ def _required_count(info, name, source):
     return int(value)
 
 
-def _validated_table(
-        connection, table_name, source_schema, options, source):
-    table = _get_or_create_table(
-        connection, table_name, source_schema, options)
-    target_schema = _target_schema(table.raw_table)
-    _validate_lerobot_schema(
-        source_schema, target_schema, source.path)
-    _strict_lerobot_table(
-        pa.Table.from_batches([], schema=source_schema),
-        target_schema,
-        source,
-        0,
-    )
-    return table
-
-
-def _get_or_create_table(connection, table_name, schema, options):
+def _create_target_table(
+        connection, table_name, source_schema, options):
+    create_options = dict(options or {})
+    managed_options = _managed_table_options(
+        connection._identifier(table_name))
+    reserved_options = set(managed_options).intersection(create_options)
+    if reserved_options:
+        raise ValueError(
+            "%s are managed by load_from_lerobot."
+            % sorted(reserved_options))
+    create_options.update(managed_options)
     try:
-        return connection.get_table(table_name)
-    except (DatabaseNotExistException, TableNotExistException):
-        return connection.create_table(
+        table = connection.create_table(
             table_name,
-            schema=schema,
-            options=options,
+            schema=source_schema,
+            options=create_options,
         )
+    except TableAlreadyExistException as error:
+        raise ValueError(
+            "LeRobot target %s already exists; use a new target table."
+            % connection._identifier(table_name)
+        ) from error
+    return table