[python] Preserve LeRobot recording metadata in direct writer

docs/docs/pypaimon/multimodal-api.mdx

@@ -702,6 +702,14 @@ same user feature mapping that would be passed to `LeRobotDataset.create`.
 The writer adds the standard `timestamp`, `frame_index`, `episode_index`,
 `index`, and `task_index` features itself.
 
+The writer preserves native LeRobot recording metadata in a managed Paimon
+table group. The named table stores frames, while `<table>__episodes`,
+`<table>__tasks`, `<table>__info`, and `<table>__stats` store episode
+boundaries, task labels, JSON-encoded dataset information, and global
+statistics. When `subtask_index` is declared, `<table>__subtasks` stores its
+ordered text vocabulary. The root table's managed options identify these
+companions.
+
 ```python
 from pypaimon.multimodal.lerobot import PaimonLeRobotWriter
 
@@ -710,6 +718,8 @@ writer = PaimonLeRobotWriter(
     "robot_data",
     fps=30,
     features=dataset_features,
+    # Required for a new table when features includes subtask_index:
+    # subtasks=["approach object", "grasp object"],
 )
 
 # LeRobot's record_loop only needs writer.fps, writer.features, and
@@ -725,10 +735,23 @@ writer.finalize()
 Like native LeRobot, `add_frame` requires every declared user feature plus a
 string `task`, and rejects caller-provided generated fields. Numeric features
 must be NumPy arrays (Torch tensors are converted) with the declared dtype and
-shape. Image metadata uses `(channels, height, width)`; image values may be CHW,
-HWC, or PIL. Images are encoded as PNG bytes and stored in Paimon `BLOB`
-columns; no LeRobot data directory or MP4 is created. `video` features remain
-unsupported.
+shape. Image dimension names may declare CHW or native HWC layout; image values
+may be CHW, HWC, or PIL. Images are encoded as PNG bytes and stored in Paimon
+`BLOB` columns; no LeRobot data directory or MP4 is created. `video` features
+remain unsupported. Task text is stored once in the tasks table; frame rows
+retain only `task_index`. Optional subtask text is likewise stored once in the
+subtasks table, while each frame supplies its declared NumPy `subtask_index`.
+Subtask labels must be non-empty and unique, and frame indices must reference
+that ordered vocabulary. On resume, the writer restores the vocabulary from
+the existing table; an explicitly supplied vocabulary must match it exactly.
+
+When `save_episode()` accepts an episode, the writer uses LeRobot's native
+statistics implementation to calculate `min`, `max`, `mean`, `std`, `count`,
+`q01`, `q10`, `q50`, `q90`, and `q99` for every non-string feature. Flattened
+episode statistics are appended to the episodes table. Image statistics use
+the encoded PNG frames and LeRobot's sampling, downsampling, CHW, and `[0,1]`
+normalization rules. Each flush aggregates the accepted episode statistics and
+replaces the global stats table; it does not rescan frame data.
 
 `save_episode` accepts the current episode and writes it to a long-lived Paimon
 batch writer. The default `episodes_per_commit=-1` keeps all completed episodes
@@ -743,16 +766,26 @@ accepts an episode, `clear_episode_buffer()` no longer affects it, even when the
 batch has not yet been committed. `finalize()` rejects an unfinished episode
 instead of silently dropping its frames.
 
-The writer creates a missing table and appends to an existing compatible table.
-Before writing, it requires the table columns, order, Arrow types, nullability,
-and LeRobot feature metadata to match. On resume, new `index` and
-`episode_index` values continue after the existing maxima, existing task
-mappings are retained, and the episode-local `frame_index` starts again at zero.
-Each commit records the next global indices and task mapping in Snapshot
-properties. Resume normally reads only that metadata; a non-empty table created
-before these properties existed is scanned once and upgraded by its next
-commit. A commit exception has an unknown result and is not automatically
-retried.
+The writer creates a missing table group and appends to an existing compatible
+group. Before writing, it requires the root columns, order, Arrow types,
+nullability, LeRobot feature metadata, managed options, companion schemas, and
+component counts to match. A legacy frame-only writer table is not implicitly
+migrated. An existing group must have the stats companion and must not have a
+subtasks companion unless its frame schema declares `subtask_index`. On resume,
+new `index` and `episode_index` values continue after the published metadata,
+existing task and subtask mappings come from their companion tables, and the
+episode-local `frame_index` starts again at zero. Snapshot properties retain
+only the next global frame and episode indices; resume restores global
+statistics from the stats table and does not scan frame data.
+
+When configured, the first `flush()` writes the immutable subtask vocabulary.
+Each flush appends new tasks and episodes, replaces global stats and info, and
+commits frames last. Paimon
+does not provide a transaction across these tables. A component commit
+exception leaves the group result unknown, makes the writer terminal, and is
+not automatically retried. Reopening validates the component state and rejects
+a partial batch. Pause writes and use `create_lerobot_tag` before training to
+pin one named snapshot on every component.
 
 ## Overwrite
 

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

@@ -51,6 +51,10 @@
     pa.field("task_index", pa.int64(), nullable=False),
     pa.field("task", pa.string(), nullable=False),
 ])
+_EMPTY_SUBTASKS_SCHEMA = pa.schema([
+    pa.field("subtask_index", pa.int64(), nullable=False),
+    pa.field("subtask", pa.string(), nullable=False),
+])
 _EMPTY_EPISODES_SCHEMA = pa.schema([
     pa.field("episode_index", pa.int64(), nullable=False),
     pa.field("dataset_from_index", pa.int64(), nullable=False),
@@ -310,6 +314,32 @@ def _append_arrow(table, data):
     return _append_arrow_tables(table, [data])
 
 
+def _overwrite_arrow(table, data):
+    target_schema = _target_schema(table)
+    if not data.schema.equals(target_schema, check_metadata=False):
+        raise ValueError(
+            "LeRobot component schema %s does not match target %s."
+            % (data.schema, target_schema))
+    builder = table.new_batch_write_builder().overwrite()
+    table_write = builder.new_write()
+    table_commit = builder.new_commit()
+    commit_started = False
+    try:
+        table_write.write_arrow(data)
+        messages = table_write.prepare_commit()
+        commit_started = True
+        table_commit.commit(messages)
+    except BaseException:
+        if not commit_started:
+            table_write.abort()
+        raise
+    finally:
+        try:
+            table_write.close()
+        finally:
+            table_commit.close()
+
+
 def _append_arrow_tables(table, tables):
     builder = table.new_batch_write_builder()
     table_write = None