Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 49 additions & 16 deletions docs/docs/pypaimon/multimodal-api.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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={
Expand All @@ -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.
Expand Down
20 changes: 16 additions & 4 deletions paimon-python/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
15 changes: 13 additions & 2 deletions paimon-python/pypaimon/multimodal/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
36 changes: 22 additions & 14 deletions paimon-python/pypaimon/multimodal/lerobot/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
"""
Expand All @@ -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")
Expand All @@ -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,
Expand All @@ -109,6 +116,7 @@ def load_from_lerobot(
batch_size,
options,
metadata,
tag_name,
)
finally:
close = getattr(dataset, "close", None)
Expand All @@ -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),
Expand All @@ -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):
Expand Down Expand Up @@ -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."
Expand Down
Loading
Loading