[python] Add RoboMIND AgileX HDF5 pipeline

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

@@ -141,6 +141,7 @@ jobs:
             python -c "import h5py; print('h5py', h5py.__version__)"
 
             if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
+              # Run the RoboMIND pipeline tests with synthetic local HDF5 data.
               # Exercise the 0.4 API in one lane until its wheel is published.
               python -m pip install "git+https://github.com/apache/paimon-rust.git@${PYPAIMON_RUST_REV}#subdirectory=bindings/python"
               python -m pip install "./paimon-python[sql]"

docs/docs/pypaimon/robomind-agilex.md

@@ -0,0 +1,134 @@
+---
+title: "RoboMIND AgileX"
+sidebar_position: 7
+---
+
+<!--
+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.
+-->
+
+# RoboMIND AgileX
+
+The RoboMIND AgileX sample turns a downloaded HDF5 directory into three
+Paimon tables:
+
+- `episodes_agilex` stores episode metadata derived from the dataset layout;
+- `frames_agilex` stores ordered robot state, raw action, RGB, and depth rows;
+- `feature_stats_agilex` versions the train-split statistics consumed by
+  policy training.
+
+The local and Ray paths use the same `RoboMindAgileXEpisodeTransform` and
+`RoboMindAgileXFrameTransform` contracts and table schemas. Ray assigns each
+complete HDF5 file to one transform task, while the Paimon sink performs one
+coordinated commit. Discovery does not open or hash HDF5 contents; validation
+and frame counting happen after a transform task has opened the file.
+
+`split` is RoboMIND dataset metadata derived from the `train` or `val`
+directory component. It is not a required field of every Paimon multimodal
+table. This sample uses successful train episodes to select frame rows for
+normalization statistics.
+
+## Run the local pipeline
+
+After downloading RoboMIND, use Python 3.11 or later, install the HDF5 and
+Vortex extras, and provide the source and warehouse directories to one command:
+
+```bash
+pip install 'pypaimon[hdf5,vortex]'
+python -m pypaimon.sample.robomind_agilex \
+  --input /data/RoboMIND/h5_agilex_3rgb \
+  --warehouse /data/warehouse
+```
+
+The command discovers and validates every `**/data/trajectory.hdf5`, ingests
+the episode and frame tables locally, materializes the canonical action, and
+writes versioned train-split normalization statistics. It prints a JSON result
+with row counts and committed snapshot IDs. The input must already be present
+locally; the command does not download RoboMIND or contact Hugging Face.
+
+Use a new warehouse for each ingestion run. Ingestion is append-only, so
+repeating the same input against existing tables would create duplicate rows.
+Canonical-action backfill is independently retryable after schema creation.
+
+The current Hugging Face example data includes `language_raw` and
+`language_distilbert`, but these datasets are not part of the published AgileX
+HDF5 schema. The episode transform therefore validates and stores them when
+present, and writes null instruction metadata when they are absent.
+
+Pytest generates several small HDF5 episodes with the real AgileX field names,
+shapes, dtypes, split layout, and success layout, so the default test needs no
+download. To exercise a downloaded customer dataset explicitly, run:
+
+```bash
+pytest -q pypaimon/tests/robomind_agilex_pipeline_test.py \
+  --robomind-agilex-input /data/RoboMIND/h5_agilex_3rgb
+```
+
+## Python API
+
+```python
+from pypaimon.sample.robomind_agilex import (
+    backfill_canonical_action,
+    ingest_local,
+    ingest_ray,
+    run_local_pipeline,
+)
+
+# Run local ingestion and canonical-action backfill together.
+pipeline = run_local_pipeline(
+    "/data/RoboMIND/h5_agilex_3rgb",
+    "/data/warehouse",
+)
+
+# Or compose the lower-level operations explicitly. Ray chooses distributed
+# task placement; concurrency is only an optional upper bound.
+ingest = ingest_ray(
+    "/data/RoboMIND/h5_agilex_3rgb",
+    "/data/warehouse",
+    concurrency=8,
+)
+
+backfill = backfill_canonical_action(
+    "/data/warehouse",
+    statistics_version="robomind-agilex-joint-position@1",
+)
+```
+
+Episode and frame ingestion commit separately and use the generic
+`pypaimon.ray.load_from_hdf5` API in Ray mode. Canonical action materialization
+and statistics refresh also commit separately. If statistics need to be
+regenerated, call `refresh_action_statistics` without repeating ingestion or
+the row-id update.
+
+The canonical `action` is `float32(concat(master/joint_position_left,
+master/joint_position_right))`. The backfill materializes only this consumed
+14-dimensional column. It does not materialize normalized actions. Instead,
+the stats table stores the train-only population mean and standard deviation,
+the `1e-2` standard-deviation floor, the train split manifest digest, and the
+source `frames_agilex` snapshot. A training reader normalizes `action` at read
+time with that versioned row.
+
+The tables are non-primary-key append tables. Repeating ingestion therefore
+appends duplicate rows by design; it does not mean row-level update/delete is
+disabled. The sample keeps deletion vectors enabled, stores vectors with
+Vortex, and sets `blob-as-descriptor=false` because its transforms emit raw
+image/depth bytes rather than external BLOB descriptors. Parquet data format,
+dynamic bucket mode, and global-index search mode are inherited defaults and
+are not repeated in the sample options.
+
+Run local and Ray modes against separate new warehouses when comparing them.

paimon-python/conftest.py

@@ -24,6 +24,13 @@
 _force_native_for_test = False
 
 
+def pytest_addoption(parser):
+    parser.addoption(
+        "--robomind-agilex-input",
+        help="Downloaded RoboMIND AgileX directory for the optional sample test.",
+    )
+
+
 def _native_plan_enabled():
     return os.environ.get(_NATIVE_PLAN_ENV) == "1"
 

paimon-python/pypaimon/multimodal/hdf5.py

@@ -72,10 +72,14 @@ def stem(self) -> str:
 
 @dataclass(frozen=True)
 class Hdf5LoadResult:
-    """Counts and optional snapshot for one ``load_from_hdf5`` call."""
+    """Counts and optional snapshot for one ``load_from_hdf5`` call.
+
+    ``batch_count`` is unavailable for Ray loads because counting lazy output
+    batches would execute the transform a second time.
+    """
 
     file_count: int
-    batch_count: int
+    batch_count: Optional[int]
     row_count: int
     snapshot_id: Optional[int]
 
@@ -188,33 +192,17 @@ def _load_hdf5_files(table, files, transform, source_file_io, h5py):
         table_commit.add_commit_callback(snapshot_recorder)
 
         for source in files:
-            source_row_count = 0
-            with closing(source_file_io.new_input_stream(source.path)) as stream:
-                _require_seekable(stream, source)
-                with h5py.File(stream, "r") as h5:
-                    transformed = transform(h5, source)
-                    batches = None
-                    try:
-                        batches = _arrow_batches(transformed)
-                        for value in batches:
-                            arrow_table = _strict_arrow_table(
-                                value,
-                                target_schema,
-                                source,
-                                batch_count,
-                            )
-                            batch_count += 1
-                            row_count += arrow_table.num_rows
-                            source_row_count += arrow_table.num_rows
-                            if arrow_table.num_rows:
-                                table_write.write_arrow(arrow_table)
-                    finally:
-                        _close_transform_iterator(
-                            batches if batches is not None else transformed)
-
-            if source_row_count == 0:
-                raise ValueError(
-                    "HDF5 source %s produced no rows." % source.path)
+            for arrow_table in _transform_hdf5_file(
+                    source,
+                    transform,
+                    source_file_io,
+                    h5py,
+                    target_schema,
+                    batch_index=batch_count):
+                batch_count += 1
+                row_count += arrow_table.num_rows
+                if arrow_table.num_rows:
+                    table_write.write_arrow(arrow_table)
 
         commit_messages = table_write.prepare_commit()
         commit_started = True
@@ -241,6 +229,35 @@ def _load_hdf5_files(table, files, transform, source_file_io, h5py):
                 table_commit.close()
 
 
+def _transform_hdf5_file(
+        source,
+        transform,
+        source_file_io,
+        h5py,
+        target_schema,
+        *,
+        batch_index=0):
+    """Yield validated Arrow tables for one HDF5 source."""
+    produced_rows = 0
+    with closing(source_file_io.new_input_stream(source.path)) as stream:
+        _require_seekable(stream, source)
+        with h5py.File(stream, "r") as h5:
+            transformed = transform(h5, source)
+            batches = None
+            try:
+                batches = _arrow_batches(transformed)
+                for index, value in enumerate(batches, start=batch_index):
+                    arrow_table = _strict_arrow_table(
+                        value, target_schema, source, index)
+                    produced_rows += arrow_table.num_rows
+                    yield arrow_table
+            finally:
+                _close_transform_iterator(
+                    batches if batches is not None else transformed)
+    if produced_rows == 0:
+        raise ValueError("HDF5 source %s produced no rows." % source.path)
+
+
 def _discover_hdf5_files(paths, source_file_io):
     values = _path_values(paths)
     normalized = {}

paimon-python/pypaimon/ray/__init__.py

@@ -31,6 +31,7 @@
 from pypaimon.ray.update_by_row_id import update_by_row_id
 from pypaimon.ray.read_by_row_id import read_by_row_id
 from pypaimon.ray.process_row_id_ranges import process_row_id_ranges
+from pypaimon.ray.hdf5 import load_from_hdf5
 
 __all__ = [
     "read_paimon",
@@ -42,6 +43,7 @@
     "update_by_row_id",
     "read_by_row_id",
     "process_row_id_ranges",
+    "load_from_hdf5",
     "WhenMatched",
     "WhenNotMatched",
     "source_col",