[python] Add RoboMIND AgileX HDF5 pipeline

[YannByron]

Summary

Add a RoboMIND AgileX sample that uses the same HDF5 transform contract for local and Ray ingestion. The pipeline writes episode and frame tables, materializes canonical actions, and refreshes train-split statistics independently.

Changes

• Add pypaimon.ray.load_from_hdf5 with shared HDF5 discovery, transform validation, and a single coordinated Paimon commit.
• Add separate RoboMIND AgileX episode and frame transforms for local and Ray ingestion.
• Keep deletion vectors and Vortex vector storage enabled while storing image and depth payloads as raw bytes.
• Split canonical action materialization from statistics refresh and process frame updates by Paimon split.
• Generate multi-episode AgileX HDF5 fixtures in pytest, with an optional flag for downloaded customer data.
• Document the dataset layout, split semantics, table options, local usage, and Ray usage.

Testing

• python -m pytest -q pypaimon/tests/multimodal_hdf5_test.py pypaimon/tests/ray_hdf5_test.py pypaimon/tests/robomind_agilex_pipeline_test.py (29 passed, 1 skipped)
• Ruff on all changed Python files
• git diff --check origin/master...HEAD
• Customer dataset test with --robomind-agilex-input (requires a downloaded RoboMIND dataset)

Notes

This builds on the HDF5 DataSource ingestion merged in #9411. Repeating ingestion appends duplicate rows by design; a future merge-into mode can provide episode-level upserts.

[JingsongLi]

Inline findings on the current PR head.

[JingsongLi]

[P1] Read language metadata from the published sidecar annotations

The published AgileX trajectory.hdf5 layout contains the master, puppet, and observations datasets, but not language_raw or language_distilbert; RoboMIND publishes task-language annotations separately in CSV/JSON. Following the documented --input .../h5_agilex_3rgb path therefore fails on the first file here. The synthetic fixture masks this by adding both datasets itself. Please load/derive the instruction from the official sidecar/path metadata and make the embedding external or optional, then cover the actual published HDF5 shape in a test. References: official AgileX HDF5 layout, RoboMIND dataset card.

[YannByron]

Thanks. The current Hugging Face example_data we downloaded does contain both language_raw and language_distilbert, but I agree that neither field is part of the published AgileX HDF5 schema. Commit 09fc963 now treats them as optional: present datasets remain strictly shape-validated, while absent datasets produce null instruction metadata. I also added a test using the published HDF5 shape without either language dataset.

[JingsongLi]

Re-reviewed the fixes on the latest head. The previous findings are addressed, but the new batched row-id update API introduces one correctness regression.

[JingsongLi]

[P1] Coalesce updates that target the same file

Calling update_columns independently for each input batch is only safe when the batches map to disjoint first_row_id groups. If two batches update different rows in the same base file, each call rereads the same snapshot baseline and writes a full delta for the same write_cols and first_row_id; after both messages are committed, one batch's changes are lost. I reproduced this on the current head with a real batch table: batch 1 updates row 0 from age 25 to 26, batch 2 updates row 1 from 30 to 31, and the committed table reads [26, 30, 35, 40, 45] instead of [26, 31, 35, 40, 45]. The new mock-only test does not exercise file generation or reads, so it misses this. Please coalesce updates per target first_row_id and write one delta per target file (or explicitly reject batches whose target groups overlap), and add this same-file end-to-end case.

[YannByron]

Thanks for reproducing this. In the current local RoboMIND backfill path, updates come directly from Paimon planned splits. DataEvolutionSplitGenerator groups files by first_row_id before packing splits, and _iter_raw materializes each complete split as one batch, so the same first_row_id group is not expected to cross batches in this specific flow.

However, you are right that the local batch API neither expresses nor enforces this precondition, so arbitrary batches can hit this correctness issue. I also overlooked the existing Ray implementations. The public Ray update_by_row_id API already determines first_row_id before writing, shuffles and groups updates by it, coalesces all updates for the same target file in one worker, and performs one coordinated commit. RoboMIND tables enable deletion vectors, so their same-table backfill should instead reuse the existing optimized Ray self-merge path, which reads live rows from a pinned snapshot and schedules one update task per target file group. Therefore, no separate distributed update engine is needed.

For this PR, I will add exact overlap detection for the affected first_row_ids in the local batch API and fail before commit, aborting staged files instead of silently losing updates. I will also add the same-file local end-to-end test and distributed multi-block same-file coverage, and wire the RoboMIND Ray backfill to the existing self-merge path.

[JingsongLi]

+1