lance: migration plan, planner and Ray driver by RyanPCo · Pull Request #752 · GaTech-RL2/EgoVerse

RyanPCo · 2026-10-30T19:04:35Z

The plan and tooling for migrating every non-deleted episode into one Lance root, with the
robot embodiments now routed.

docs/zarr_to_lance_migration.md (force-added; docs/ is gitignored): inventory,
destination layout and recording ids, sharding and work units, worker flow, how
annotations migrate, idempotency, validation gates, cost, rollout, and the raw path for
Aria and EVA. It also notes that MANO weights (external_ckpts/, gitignored) must ship
to the workers.
egomimic/scripts/lance_pilot/plan_zarr_migration.py: read-only planner (SQL selects, R2
lists). It writes manifest.parquet (route, converter, recording_id, steps table, shard,
work unit), rejected.parquet with every reason, an inventory and a summary. Aria and EVA
rows migrate from raw only: with --raw-listing, a row whose capture is complete on R2
gets the raw route and the rest are rejected (no_raw / incomplete_raw). yam_bimanual
(EVA_STEPS, parallel_jaw_gripper) and dexmate_bimanual (DEXMATE_STEPS, dex_hand) are no
longer rejected for want of a schema.
egomimic/scripts/zarr_to_lance_ray.py: Ray driver for the zarr route, one task per
episode, resumable from committed ids, staged meta.json files and its report; a task
whose worker dies is reported as failed and the run goes on. --direct-videos uploads
MP4s straight to the Lance root (commit.stage_direct).
eva_to_lance takes lab, operator, task, scene and objects from the episode's SQL row
(--manifest), as the committer does for Aria, instead of the capture's _metadata.json;
the converter's required-field check stays.

Planner run v2 (/coc/flash7/scratch/lance_pilot/migration_plan/v2/, SQL read 2026-10-30):
289,734 episodes (514M frames) convertible into 23 steps tables and 4,539 work units.
Zarr route 284,402 (microagi v2 186,188; mecka 41,611; abc yam 36,882; scale 17,090; rl2
yam 2,334; yam 182; sharpa_dex dexmate 115). Raw route 5,332 (aria 2,569, eva 2,763).
335,892 rejected: missing_operator 335,751 (microagi v1 and trace, skipped for now),
no_zarr_path 128, missing_task 8, unknown_embodiment 5, incomplete_raw 4, no_frames 9,
zarr_processing_error 2. 99 are already in lance_v2.

Testing: on a compute node, tests/unit/test_plan_zarr_migration.py (ids, tables, robot
routing, rejections, raw index and raw-only routing) and test_eva_to_lance (SQL row
replaces the capture metadata; a null SQL operator is not filled in): 25 passed. Full unit
suite at the top of the stack, in 4 parallel jobs: 1,565 passed, 5 skipped (3 of the skips are test_mano_fit without weights in the checkout; it passes with them linked in).

🤖 Generated with Claude Code

https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw

RyanPCo · 2026-10-30T19:20:49Z

+COLUMNS = (
+    "episode_hash, operator, lab, embodiment, rig_name, scene, task, task_description,"
+    " objects, license, num_frames, zarr_processed_path, zarr_mp4_path,"
+    " zarr_processing_error, is_eval"


medium (dead-code): COLUMNS selects is_eval, which nothing reads, but it leaves out is_deleted. load_rows writes this frame to --snapshot and on the next run evaluates df[~df.is_deleted...], which raises AttributeError, so a snapshot written by this script can never be read back. Either select is_deleted (and drop is_eval), or drop the filter on the snapshot path since the SQL query already excludes deleted rows.

RyanPCo · 2026-10-30T19:20:52Z

+def load_rows(snapshot: Path | None) -> pd.DataFrame:
+    if snapshot and snapshot.exists():
+        df = pd.read_parquet(snapshot)
+        return df[~df.is_deleted.astype(bool)].reset_index(drop=True)


medium (correctness): --snapshot can't be read back. COLUMNS doesn't select is_deleted, so the parquet written on the first run has no such column, and on the second run df.is_deleted raises AttributeError. The SQL query already filters not is_deleted, so just return pd.read_parquet(snapshot) here (or add is_deleted to COLUMNS).

RyanPCo · 2026-10-30T19:20:54Z

+    if args.lance:
+        m = m[~m.recording_id.isin(commit.committed_ids(args.lance))]
+    if args.stage_to:
+        staged = set(commit.list_dirs(commit.join(args.stage_to, STAGING_SOURCE)))


medium (data-integrity): A delivery whose upload died partway (worker killed, or put_file raised, so status failed) leaves <stage-to>/zarr/<id>/ without meta.json. This check still treats it as staged, even with --retry-failed, and the committer ignores it because meta.json is missing, so the episode never migrates. Count an id as staged only if <id>/meta.json exists.

RyanPCo · 2026-10-30T19:20:56Z

+            if not pending:
+                break
+            finished, pending = ray.wait(pending, num_returns=1)
+            res = ray.get(finished[0])


medium (error-handling): If a task dies outside Python (OOM kill, worker crash after max_retries), ray.get raises here and the whole driver exits, losing every in-flight task. No report row is written for that episode, so a resume picks it again and can crash the same way. Wrap ray.get in try/except, keep a ref-to-row map, and write a failed row with repr(e). Also guard the select() report read against an empty file: an empty pd.read_json(..., lines=True) has no episode_hash column.

RyanPCo · 2026-10-30T19:20:58Z



-def convert(args) -> dict:
+def _value(v):


low (duplication): _value and _objects repeat zarr_to_lance._text and zarr_to_lance._objects. The only difference is that _value skips str(). Move one text_or_none/objects_list pair into a shared module (or reuse zarr_to_lance's) so the raw-EVA and zarr routes can't drift apart on how they parse SQL objects.

RyanPCo · 2026-10-30T19:21:00Z

+    return {**md, **{k: _value(sql.get(k)) for k in SQL_FIELDS}}
+
+
+def read_manifest(path: Path) -> dict[str, dict]:


low (duplication): There are now two read_manifest functions. zarr_to_lance.read_manifest reads CSV only (string values), and this one reads parquet or CSV through pandas (typed values, NaN for blanks). The planner writes manifest.parquet, so zarr_to_lance --manifest can't read its output. Keep one reader (this pandas version) in a shared place and import it from both converters.

RyanPCo · 2026-10-30T19:21:02Z

+
+
+# Values zarr_to_lance.episode_metadata rejects for lab, task_name and operator.
+PLACEHOLDERS = {"", "none", "null", "nan", "unknown", "debug"}


low (duplication): PLACEHOLDERS and placeholder copy zarr_to_lance.PLACEHOLDERS and _placeholder, and the comment says they have to match. Import them from zarr_to_lance (make _placeholder public) so the planner can't route episodes that the converter will then reject.

RyanPCo · 2026-10-30T19:21:04Z

+
+    sample = None
+    if args.check_r2:
+        both = pd.concat([ok, bad])


low (correctness): ok and bad both have indexes starting at 0, so this concat has duplicate labels. rows.loc[picks] in check_r2 then returns rows from both frames for each label, often from other strata, which inflates and mixes the R2 sample and skews bytes_per_frame, est_zarr_gb and sample_missing_frac. Use pd.concat([ok, bad], ignore_index=True).

RyanPCo · 2026-10-30T19:21:06Z

+    return ok, bad
+
+
+def _list(uri: str) -> tuple[int, int]:


nit (duplication): _list repeats the pagination in commit._objects. You could write files = commit.list_files(uri); return len(files), sum(s for s, _ in files.values()). download_prefix in zarr_to_lance_ray could iterate commit.list_files too.

RyanPCo · 2026-10-30T19:21:10Z

+    return df
+
+
+def _epoch_ms(episode_hash) -> str | None:


nit (conventions): aws_sql is already imported at the top, so import episode_hash_to_timestamp_ms there as well. Catch only (ValueError, TypeError) so that an unexpected error isn't quietly treated as a non-timestamp hash and routed to the zarr namespace.

RyanPCo · 2026-10-30T19:21:12Z

+
+log = logging.getLogger("zarr_to_lance_ray")
+
+STAGING_SOURCE = "zarr"


nit (duplication): STAGING_SOURCE repeats zarr_to_lance.STAGING_SOURCE. The staging dir used here and in select() has to match the converter's, so import it instead of redefining it.

RyanPCo · 2026-10-30T19:21:14Z

+            work / "out",
+            metadata,
+            episode_hash=row["episode_hash"],
+            **({"ffmpeg": opts["ffmpeg"]} if opts.get("ffmpeg") else {}),


nit (simplification): Give --ffmpeg the default zarr_to_lance.FFMPEG (or "ffmpeg"), as the converter's own CLI does, then pass ffmpeg=opts["ffmpeg"] directly instead of the conditional **{...} splat.

RyanPCo

Automated adversarial review (read-only; findings were checked by a second skeptical pass before posting): 4 medium, 4 low, 5 nit.

Comments that could not be anchored to a diff line:

egomimic/scripts/eva_process/eva_to_lance.py:82 nit (simplification): This copies zarr_to_lance.PLACEHOLDERS. If the two drift, the planner will accept rows the converter rejects, or the reverse. Import it from egomimic.scripts.zarr_to_lance (or a shared module), and likewise reuse _objects in eva_to_lance instead of redefining it.

RyanPCo · 2026-10-30T19:41:55Z

Review follow-up (force-pushed). 12 of 13 findings fixed fully and 1 in part:

Half-staged deliveries: the driver counts an episode as staged only when <stage-to>/zarr/<id>/meta.json exists (new test).
A crashed task (OOM or dead worker) is caught around ray.get and recorded as failed for its row, and the run continues. An empty report file is guarded (new test).
--snapshot reads back: the snapshot path no longer filters on the unselected is_deleted, since the query already excludes deleted rows (new test).
is_eval dropped from the planner's columns.
check_r2 input is built with pd.concat(..., ignore_index=True).
Shared helpers: eva_to_lance's _value/_objects/read_manifest and the planner's PLACEHOLDERS/placeholder now come from egomimic/rldb/lance/metadata.py, added in zarr_to_lance: vendor and robot zarr episodes into Lance, with their annotations #751. One read_manifest (parquet or CSV) now serves both converters.
_list and download_prefix use commit.list_files.
STAGING_SOURCE is imported from zarr_to_lance in the driver and the planner. --ffmpeg defaults to zarr_to_lance.FFMPEG.
_epoch_ms (partly fixed): the import moved to the top and only ValueError is caught. Not done: computing the ms column once. It costs one strptime per row in two passes, which is negligible next to the SQL read, and caching it would add a column threaded through plan.
The docs describe the new GC, resume and replace behavior.

https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw

The committer takes Aria's recording and task metadata from SQL, but eva_to_lance read them from the capture's _metadata.json, so the two raw routes could disagree with the episodes table. convert() now takes the SQL row (--manifest: a migration manifest or an app.episodes export) and its lab, operator, task, task_description, scene and objects replace the capture's; a field the row leaves null stays null and the required check rejects it. Without a row the capture's values are used as before. Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw

docs/zarr_to_lance_migration.md plans converting every non-deleted episode into one Lance root: inventory, destination layout and ids, sharding, worker flow, idempotency, validation gates, cost, rollout, and the raw path for Aria and EVA. plan_zarr_migration reads app.episodes (read-only) and writes a manifest of convertible episodes with their route, converter, recording_id, steps table and work unit, plus the rejected rows with every reason. Aria and EVA migrate from raw only: with --raw-listing, rows whose capture is complete on R2 get the raw route and the rest are rejected (no_raw, incomplete_raw). yam_bimanual (EVA_STEPS, parallel_jaw_gripper) and dexmate_bimanual (DEXMATE_STEPS, dex_hand) are now routed like the vendor zarrs. zarr_to_lance_ray converts and stages the zarr-route rows, one Ray task per episode, resumable from committed ids, staged meta.json files and its report; a task whose worker dies is reported as failed and the run goes on. --direct-videos uploads the MP4s to the Lance root and stages only steps and meta.json. Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw

RyanPCo commented Oct 30, 2026

View reviewed changes

RyanPCo force-pushed the ryanco/zarr-to-lance branch from eab64d3 to d5864a2 Compare October 30, 2026 19:41

RyanPCo force-pushed the ryanco/lance-migration-plan branch from 72692a6 to 369caa4 Compare October 30, 2026 19:41

RyanPCo added 2 commits October 30, 2026 18:55

RyanPCo force-pushed the ryanco/zarr-to-lance branch from d5864a2 to 853c113 Compare October 30, 2026 23:05

RyanPCo force-pushed the ryanco/lance-migration-plan branch from 369caa4 to 01cbd79 Compare October 30, 2026 23:05

		return {md, {k: _value(sql.get(k)) for k in SQL_FIELDS}}


		def read_manifest(path: Path) -> dict[str, dict]:



		# Values zarr_to_lance.episode_metadata rejects for lab, task_name and operator.
		PLACEHOLDERS = {"", "none", "null", "nan", "unknown", "debug"}


		log = logging.getLogger("zarr_to_lance_ray")

		STAGING_SOURCE = "zarr"

Conversation

RyanPCo commented Oct 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RyanPCo left a comment

Choose a reason for hiding this comment

Uh oh!

RyanPCo commented Oct 30, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

RyanPCo commented Oct 30, 2026 •

edited

Loading