Repository navigation
lance migration - #716
lance migration#716RyanPCo wants to merge 6 commits into
Conversation
cbe2e50 to
de7997e
Compare
RyanPCo
changed the title
651e44b to
3e9f77a
Compare
2f40965 to
e890bcb
Compare
e890bcb to
136fbff
Compare
7677f76 to
648a2a7
Compare
121ad91 to
49123e8
Compare
RyanPCo
left a comment
RyanPCo
left a comment
There was a problem hiding this comment.
Automated adversarial review (read-only; findings were checked by a second skeptical pass before posting): 1 high, 6 medium, 10 low, 11 nit.
| out = [None] * len(idxs) | ||
| groups: dict[type, list] = {} | ||
| for pos, idx in enumerate(idxs): | ||
| name, local = self.index_map[idx] |
There was a problem hiding this comment.
high (correctness): EvenStrideDataset inherits this getitems without overriding it. Inside it, self.index_map resolves through __getattr__ to base.index_map, so a stride-space index (0..len(indices)) is read as a base-global index. For LanceStreamingDataset leaves (the only ones with fetch_planned), a strided val loader silently evaluates roughly the first len(indices) base samples instead of the strided subset. Please override it in EvenStrideDataset: def __getitems__(self, idxs): return self.base.__getitems__([self.indices[i] for i in idxs]), and add a test that compares batched and per-item output for a wrapped streaming dataset.
|
|
||
|
|
||
| def recording_id(source: str, vendor_recording_id: str) -> str: | ||
| """`source` is a fixed per-converter namespace ("aria", "eva"), unrelated to the `lab` column.""" |
There was a problem hiding this comment.
medium (consistency): This docstring says the first argument is a per-converter namespace ("aria") unrelated to lab. mcap/spec.py (which says it copies this function), validate.py and SPEC.md all call it with the lab. Both use the same UUIDv5 namespace, so one vendor recording gets uuid5(ns, "aria/<id>") from aria_to_lance and uuid5(ns, "rl2/<id>") from the validator. The promised agreement test only compares the namespace constant, so it won't catch this. Please pick one meaning and use it in this function, spec.py, SPEC.md and the aria_to_lance/duplicate.py callers.
| ) | ||
| episodes = { | ||
| h: norm_cache.episode_fingerprint(getattr(ds, "episode_path", None)) | ||
| h: norm_cache.leaf_fingerprint(ds) |
There was a problem hiding this comment.
medium (correctness): This switches the norm-cache key to norm_cache.leaf_fingerprint(ds), but egomimic/scripts/precompute_norm_stats.py (~line 173) still uses episode_fingerprint(getattr(ds, "episode_path", None)). For Lance leaves the two fingerprints differ, so a precompute run on a Lance config such as aria_pilot_lance writes an entry that training never hits, and training quietly recomputes. Switch precompute to leaf_fingerprint too, or better, move the episodes-dict construction into one norm_cache helper that both callers use.
| embodiment=text_or_none(s.embodiment), | ||
| end_effector=m.get("end_effector"), | ||
| rig_name=text_or_none(s.rig_name), | ||
| operator=text_or_none(s.operator), |
There was a problem hiding this comment.
medium (data-integrity): operator and task_name (line 126) can come out of SQL as None, and nothing rejects them: commit.steps_key only checks lab/embodiment/end_effector. Recordings with missing operator or task then commit, and task filters silently drop them. Reject the delivery when either is missing. Better, enforce required metadata (lab, task_name, operator) in rldb/lance/commit so every committer gets the check.
| def append(root: Path, deliveries: list[Delivery]) -> list[str]: | ||
| """Commit new recordings; ones already in `recordings` are skipped, so re-running is a no-op.""" | ||
| have = committed_ids(root) | ||
| todo = [d for d in deliveries if d.recording_id not in have] |
There was a problem hiding this comment.
medium (data-integrity): append() is not idempotent after a partial failure. Steps are committed before recordings, and have only reflects recordings.lance. If the process dies after the steps append, a re-run appends the same steps rows again. The same happens if deliveries contains one recording_id twice, because todo is never deduplicated. LanceEpisodeResolver._steps_rows then raises ('not contiguous' or 'steps rows != num_frames') for the whole (lab, embodiment, end_effector) table until someone deletes the rows by hand. Suggested fix: deduplicate todo by recording_id, and write steps through _replace(steps_uri, 'recording_id IN (...)', table) so orphaned rows from an earlier failed attempt get dropped. Please also add a test that simulates a crash before the recordings commit.
|
|
||
| - Lance: `storage_options` gains `allow_http` (plain-http endpoints), reading the same R2_* vars. | ||
| - zarr: `ZarrEpisode` opens s3:// paths through obstore, and reopens lazily in each worker rather | ||
| than pickling a live client (as logs/claude_scratch/zarr_r2_patch.py does for R2). |
There was a problem hiding this comment.
nit (leftover): logs/claude_scratch/zarr_r2_patch.py is not in the repo. Drop the parenthetical or describe the R2 pattern inline.
| @@ -0,0 +1,35 @@ | |||
| # Lance pilot baseline: the pilot's bimanual Aria zarrs, read locally. Pair with | |||
| # aria_pilot_lance.yaml (same recordings, keymap and transforms); see | |||
| # docs/lance_aria_pilot_plan.md. | |||
There was a problem hiding this comment.
nit (docs): docs/lance_aria_pilot_plan.md does not exist on this branch or later ones in the stack. Remove the reference or add the doc. tests/integration/test_lance_versions.py likewise refers to 'Phase 7 of the Lance pilot' and prints PHASE7; describe the test without the plan's phase numbering.
| stream_ts = {"rgb": list(rgb_ts)} | ||
|
|
||
| # Same per-frame pose math as aria_to_zarr, on every RGB frame (no clean_data). | ||
| both = "both" |
There was a problem hiding this comment.
nit (dead-code): both = "both" only aliases a literal; pass arm="both" directly. log at line 35 is never used and can be removed.
| assert _all_same(s1, _samples(t1), s1) == [] | ||
| assert {h: leaf.cache_fingerprint() for h, leaf in t1.items()} == fp1 | ||
| assert _row_count_ok(root, tag="t1") and _row_count_ok(root, tag="t6") | ||
| with pytest.raises(Exception): |
There was a problem hiding this comment.
nit (test-quality): pytest.raises(Exception) also passes on an unrelated TypeError or KeyError. Narrow it to the exception Lance raises for a missing version and add match= so the test checks what it means to check.
|
|
||
| scratch/ | ||
| sample_data/ | ||
| /fb/ |
There was a problem hiding this comment.
nit (leftover): Nothing in the repo writes to /fb/ (every bench --out is required or defaults elsewhere), so this looks like local working state. Remove it, or have the bench scripts default to an existing ignored directory such as scratch/.
Lance tables (steps, recordings, tasks, aliases) and MP4 videos for Aria recordings, a LanceEpisodeResolver that builds ZarrDataset-compatible leaves, a committer (append, replace, edit, add_columns, tag, compact, cleanup), the aria_to_lance converter, pilot benchmarks and A/B data configs. Norm caches key on a leaf's own fingerprint so Lance leaves reuse them; loaders over Lance use forkserver workers. resolve() finds recording boundaries with pc.run_end_encode instead of a numpy object array: 8.4 s -> 3.2 s at 6,120 recordings. Streaming from object storage: LanceStreamingDataset reads frames from 8-frame video chunk rows, and MultiDataset.__getitems__ fetches a whole DataLoader batch at once (one steps take plus one read_blobs per chunk table, overlapped). format_bench compares LeRobot v3.0, zarr and Lance on storage, startup and local/S3 throughput. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_0178DzpKv6gFrj6aHQUnvPUr
…; rdt read pattern Claude-Session: https://claude.ai/code/session_01AeFJsNDQrLsyrDyJ2xgbrW
… end_effector)
`lab` held the source format and `site` the collecting lab; lab now means the
collecting lab as in zarr and SQL, and site is gone, so filter lambdas see the
column directly instead of through a resolver rename. A required
`end_effector` column (END_EFFECTOR: human_hand, parallel_jaw_gripper) joins embodiment, and
steps tables are split per (lab, embodiment, end_effector) under
steps/{lab}/{embodiment}/{end_effector}.lance. The committer derives the table
from each recording and rejects missing or unknown values; the resolver reads
each recording from its own table, so one resolver can span labs. Bench and
pilot tools derive the steps path from the recordings table.
recording_id keeps hashing the converter's fixed source name ("aria"), so
existing ids and everything derived from them are unchanged.
fps=10 cases in the image history/memory tests move to 60: only 30 and 60
will be valid episode rates.
Drops aria_pilot_lance_stream.yaml: its S3 root
(lance_pilot/lance_faststart) is stale and nothing else references it.
Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw
The lance tables and loader in this PR import lance; without the pin CI cannot import them. Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw
…ithout obstore The JPEG retry lands on a random frame. When that frame is under 3, the history window repeats frame 0, but the expected slice was shorter. CI hit it on every run. obstore is only used by the format bench and is not a dependency. Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw
49123e8 to
b7ceab0
Compare
|
Review findings: status after the fixup pass. The fixes are squashed into this PR's commits, and the whole stack was force-pushed. Fixed
Partial
Not fixed
|
Lance tables (steps, recordings, tasks, aliases) and MP4 videos for Aria
recordings, a LanceEpisodeResolver that builds ZarrDataset-compatible
leaves, a committer (append, replace, edit, add_columns, tag, compact,
cleanup), the aria_to_lance converter, pilot benchmarks and A/B data
configs. Norm caches key on a leaf's own fingerprint so Lance leaves
reuse them; loaders over Lance use forkserver workers.
labis the collecting lab (e.g. rl2), as in zarr and SQL; there is nositecolumn, so filter lambdas seelabdirectly. A requiredend_effectorcolumn (END_EFFECTOR: hand, gripper) sits next toembodiment, and steps are split into one table per (lab, embodiment,
end_effector) at steps/{lab}/{embodiment}/{end_effector}.lance. The
committer derives each recording's table and rejects missing or unknown
values; the resolver reads each recording from its own table, so one
resolver can span labs. recording_id(source, vendor_recording_id) hashes
the converter's fixed source name ("aria/"), unrelated to
lab.resolve() finds recording boundaries with pc.run_end_encode instead of a
numpy object array: 8.4 s -> 3.2 s at 6,120 recordings.
Streaming from object storage: LanceStreamingDataset reads frames from
8-frame video chunk rows, and MultiDataset.getitems fetches a whole
DataLoader batch at once (one steps take plus one read_blobs per chunk
table, overlapped). format_bench compares LeRobot v3.0, zarr and Lance
on storage, startup and local/S3 throughput.
The fps=10 cases in the image history/memory tests move to 60, since only
30 and 60 will be valid episode rates.
Stacked on #689 (previously #690). Depends on pylance 13.0.0 (pyproject.toml, uv.lock).
🤖 Generated with Claude Code
https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw