[python] Add RoboMIND AgileX HDF5 pipeline - #9445
Conversation
Reuse the HDF5 transform contract for local and Ray ingestion, then materialize canonical actions and independently refresh train statistics. Co-Authored-By: Codex <noreply@anthropic.com> Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 1112/1112 AI-Contributed/UT: 342/342
Separate the module constants from the EpisodeSource class declaration so the Python CI matrix passes E302. Co-Authored-By: Codex <noreply@anthropic.com> Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0
Skip only the end-to-end ingestion tests when vortex-data is unavailable while retaining transform and statistics coverage on older Python versions. Co-Authored-By: Codex <noreply@anthropic.com> Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol Co-Authored-By: Codex <noreply@ai-tool.com> AI-Contributed/Feature: 0/0 AI-Contributed/UT: 9/9
Return exact Ray commit metadata without rescanning tables, stabilize action statistics, and strengthen validation and pipeline tests. Co-Authored-By: Codex <noreply@anthropic.com> AI-Model: gpt-5 Co-Authored-By: Codex <noreply@ai-tool.com> AI-Contributed/Feature: 204/204 AI-Contributed/UT: 193/193
JingsongLi
left a comment
There was a problem hiding this comment.
Inline findings on the current PR head.
| frame_count = lengths.pop() | ||
| if frame_count <= 0: | ||
| raise ValueError("%s: episode has no frames." % source_key) | ||
| _instruction(h5, source_key) |
There was a problem hiding this comment.
[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.
| continue | ||
| updates = build_canonical_action_backfill(source) | ||
| messages.extend( | ||
| builder.new_update() |
There was a problem hiding this comment.
[P1] Reuse one row-id file index across all splits
Each update_by_arrow_with_row_id call constructs a new TableUpdateByRowId; its constructor runs plan_for_write() and rebuilds the index of every current data file. Because that happens inside the split loop, N splits perform N full manifest/file scans, making the backfill approximately quadratic in split count. This is material for the full AgileX dataset (10,629 trajectories, generally hundreds of frames each). Please plan/index the snapshot once and reuse a batch-capable updater across the streamed splits while retaining the single final commit.
| FEATURE_STATS_TABLE = "feature_stats_agilex" | ||
| DEFAULT_STATISTICS_VERSION = "robomind-agilex-joint-position@1" | ||
|
|
||
| TABLE_OPTIONS = { |
There was a problem hiding this comment.
[P2] Pin the intended vector file format
This sample, its tests, and the documentation expect Vortex, but current master changed the multimodal default to Parquet in #9444. Since TABLE_OPTIONS does not set vector.file.format, the merged code will inherit Parquet and the Vortex assertion will fail once CI runs against the updated base. Please either set vector.file.format=vortex explicitly (and document the pypaimon[vortex] / Python >= 3.11 requirement), or accept Parquet and update the tests and documentation accordingly.
| from pypaimon.schema.data_types import AtomicType, VectorType | ||
| from pypaimon.schema.schema_change import SchemaChange | ||
|
|
||
| connection.catalog.alter_table( |
There was a problem hiding this comment.
[P2] Make the schema-change stage retryable
alter_table durably adds the nullable action column before the row-id update is committed. If update generation, commit, or the subsequent snapshot lookup fails, a retry immediately hits _validate_backfill_target and rejects the existing column, leaving the warehouse stuck with an empty or incomplete action. Please treat an existing compatible column as a resumable state and rerun the deterministic row-id update (or persist explicit stage state), and add a failure-injection retry test.
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
Testing
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.