@@ -616,32 +616,39 @@ orphan files for normal Paimon cleanup.
 ## Load LeRobot Dataset v3
 
 `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 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.
+repository. It derives the schema from `meta/info.json` and writes one row per
+frame. The import creates `<table>__episodes`, `<table>__tasks`, `<table>__info`,
+and optional `<table>__stats` and `<table>__subtasks` companion tables. Task text
+remains in the task table; frames retain `task_index`.
+
+Info and stats use `key STRING, value STRING`: each top-level info property or
+stats feature becomes one row. Each value is JSON-encoded, preserving nested
+objects, arrays, nulls, and scalar types. Decode it with `json.loads`; for example,
+`fps` stores `30`, while `codebase_version` stores `"v3.0"` (including quotes).
+Statistics may contain `NaN` and `Infinity`, supported by Python's JSON decoder.
+Missing or empty stats create no stats table. The frame table's
+`pypaimon.lerobot.<component>-table` options identify the components created.
 
 ```shell
 pip install 'pypaimon[lerobot]'
 ```
 
 ```python
-version_id = conn.load_from_lerobot(
+conn.load_from_lerobot(
     "robot_data",
     "/data/lerobot_dataset",
+    tag_name="initial-import",  # Optional: tag all imported component snapshots.
 )
-print(version_id)
 ```
 
-The returned `version_id` is the common tag name for all dataset components.
+The call returns `None` on success. Omitting `tag_name` imports the tables
+without creating tags. The one-time importer requires a new target table and
+a non-empty source dataset; subsequent table edits use the normal Paimon APIs.
 
 For FileIO URIs, pass credentials through `source_options`:
 
 ```python
-version_id = conn.load_from_lerobot(
+conn.load_from_lerobot(
     "robot_data",
     "oss://source-bucket/lerobot_dataset",
     source_options={
@@ -652,11 +659,37 @@ version_id = conn.load_from_lerobot(
 )
 ```
 
-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.
+Before training, finish any related data/metadata updates and pause writes to
+this table group. Create a common named tag over the current component snapshots:
+
+```python
+tag = "train-2026-09-07"
+snapshots = conn.create_lerobot_tag("robot_data", tag)
+frames = conn.get_table("robot_data").scan(tag_name=tag).to_arrow()
+
+# Companion tables are ordinary Paimon tables, read with the same tag.
+info_table = conn.catalog.get_table("default.robot_data__info").copy(
+    {"scan.tag-name": tag})
+builder = info_table.new_read_builder()
+info_rows = builder.new_read().to_arrow(builder.new_scan().plan().splits())
+
+import json
+info = {
+    row["key"]: json.loads(row["value"])
+    for row in info_rows.to_pylist()
+}
+```
+
+`create_lerobot_tag` returns component names mapped to snapshot IDs; these IDs
+may differ across tables. Later appends do not change tagged reads. Read every
+required component (including training statistics) through the same tag, and
+never fall back to latest if a tag is missing.
+
+Cross-table tagging is not atomic. Use a tag only after the creation call
+succeeds. A failure may leave partial tags; retry with writes still paused and
+unchanged snapshots, or choose a new name after repairing the group. Existing
+tags are never moved to different snapshots. Retain or delete component tags
+together, and keep writers paused until the call returns.
 
 Scalars map to scalar types, vectors to `VECTOR`, higher-rank tensors to nested
 `ARRAY`, and images to `BLOB`. Images keep their compressed bytes.
 
@@ -44,17 +44,29 @@ pip install 'pypaimon[lerobot]'
 import pypaimon.multimodal as pmm
 
 connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
-version_id = connection.load_from_lerobot(
+connection.load_from_lerobot(
     "robot_data",
     "/data/lerobot_dataset",
 )
-print(version_id)
 ```
 
 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`.
+Episode, task, info, and optional stats/subtask tables. Info and stats use
+`key STRING, value STRING` rows, with each value JSON-encoded to preserve
+nested metadata. Decode values with `json.loads`.
+
+Before training, pause writes and create a shared tag:
+
+```python
+connection.create_lerobot_tag("robot_data", "train-2026-09-07")
+frames = connection.get_table("robot_data").scan(
+    tag_name="train-2026-09-07").to_arrow()
+```
+
+Read every metadata component with the same tag. Use the tag only after creation
+succeeds; cross-table tagging is not atomic. Alternatively, pass `tag_name` to
+`load_from_lerobot` to tag the imported snapshots immediately.
 
 # HDF5 to multimodal tables
 
 
@@ -141,18 +141,29 @@ def load_from_lerobot(
             *,
             batch_size: int = 1024,
             options=None,
-            source_options=None):
+            source_options=None,
+            tag_name=None) -> None:
         """Import LeRobot Dataset v3 into a new Paimon table group."""
         from pypaimon.multimodal.lerobot import load_from_lerobot
-        return load_from_lerobot(
+        load_from_lerobot(
             self,
             table_name,
             source,
             batch_size=batch_size,
             options=options,
             source_options=source_options,
+            tag_name=tag_name,
         )
 
+    def create_lerobot_tag(self, table_name: str, tag_name: str):
+        """Pin all LeRobot components; pause group writes until this returns.
+
+        Returns component snapshot IDs. Use the tag only after success and
+        retain it on every component for the lifetime of a training run.
+        """
+        from pypaimon.multimodal.lerobot.metadata import create_lerobot_tag
+        return create_lerobot_tag(self, table_name, tag_name)
+
     def load_from_rosbag(
             self,
             table_name: str,
 
@@ -22,12 +22,14 @@
 
 from pypaimon.catalog.catalog_exception import TableAlreadyExistException
 from pypaimon.multimodal.lerobot.metadata import (
+    _COMPANION_OPTION_KEYS,
     _append_arrow_tables,
     _load_dataset_metadata,
     _managed_table_options,
     _prepare_metadata_tables,
     _positive_integer,
-    _publish_dataset,
+    _commit_metadata,
+    _validate_tag_name,
     _validated_episode_tables,
 )
 from pypaimon.multimodal.lerobot.loader import _write_dataset
@@ -58,11 +60,13 @@ def load_from_lerobot(
         batch_size: int = 1024,
         options: Optional[Mapping[str, object]] = None,
         source_options: Optional[Mapping[str, object]] = None,
-) -> int:
-    """Import LeRobot Dataset v3 and return its version ID.
+        tag_name: Optional[str] = None,
+) -> None:
+    """Import LeRobot Dataset v3 into a new Paimon table group.
 
     A new target table is created from LeRobot metadata. Episode, task, and
-    version metadata are stored in companion Paimon tables.
+    info/stats metadata are stored in companion Paimon tables. If provided,
+    ``tag_name`` pins all components to their imported snapshots.
     FileIO URI credentials come only from ``source_options`` and are not
     inherited from the target Catalog.
     """
@@ -74,6 +78,9 @@ def load_from_lerobot(
             or batch_size <= 0:
         raise ValueError("batch_size must be a positive integer.")
 
+    if tag_name is not None:
+        _validate_tag_name(tag_name)
+
     validated_source_options = _validated_source_options(source_options)
     _validate_source_kerberos(
         [source], validated_source_options, "LeRobot")
@@ -99,7 +106,7 @@ def load_from_lerobot(
             lerobot_schema = _schema_from_info(info)
             metadata = _load_dataset_metadata(
                 dataset, info, resolved_source)
-            return _import_dataset(
+            _import_dataset(
                 connection,
                 table_name,
                 dataset,
@@ -109,6 +116,7 @@ def load_from_lerobot(
                 batch_size,
                 options,
                 metadata,
+                tag_name,
             )
         finally:
             close = getattr(dataset, "close", None)
@@ -125,12 +133,12 @@ def _import_dataset(
         source_schema,
         batch_size,
         options,
-        metadata):
+        metadata,
+        tag_name):
     table = _create_target_table(
-        connection, table_name, source_schema, options)
+        connection, table_name, source_schema, options, metadata)
     tables = _prepare_metadata_tables(
         connection, table.raw_table, metadata)
-    version_id = 1
     episodes_snapshot_id = _append_arrow_tables(
         tables["episodes"],
         _validated_episode_tables(metadata),
@@ -146,16 +154,15 @@ def _import_dataset(
             batch_size,
             metadata,
         )
-    _publish_dataset(
+    _commit_metadata(
         connection,
         tables,
-        version_id,
+        tag_name,
         metadata,
         table.identifier,
         frames_snapshot_id,
         episodes_snapshot_id,
     )
-    return version_id
 
 
 def _validated_counts(info, source):
@@ -188,11 +195,12 @@ def _required_count(info, name, source):
 
 
 def _create_target_table(
-        connection, table_name, source_schema, options):
+        connection, table_name, source_schema, options, metadata):
     create_options = dict(options or {})
     managed_options = _managed_table_options(
-        connection._identifier(table_name))
-    reserved_options = set(managed_options).intersection(create_options)
+        connection._identifier(table_name), metadata)
+    reserved_options = set(_COMPANION_OPTION_KEYS.values()).intersection(
+        create_options)
     if reserved_options:
         raise ValueError(
             "%s are managed by load_from_lerobot."