Skip to content

lance: S3 committer fed by a staging prefix - #731

Draft
RyanPCo wants to merge 2 commits into
ryanco/lance-mp4-rangesfrom
ryanco/lance-s3-committer
Draft

RyanPCo wants to merge 2 commits into
ryanco/lance-mp4-rangesfrom
ryanco/lance-s3-committer

Conversation

@RyanPCo

@RyanPCo RyanPCo commented Oct 22, 2026 •

Copy link
Copy Markdown
Contributor

The committer (egomimic/scripts/lance_commit.py, moved from
lance_pilot/commit.py) is the only writer of the Lance tables.
Converters stage deliveries under /// with
meta.json uploaded last (commit.stage; --stage-to on aria_to_lance,
eva_to_lance and the Ray driver, which no longer commits). Each run
(lance_commit --source aria|eva) validates complete deliveries, commits
them in batches, records each batch in a new changes table and tags
every table -, then deletes their staged files. A
delivery that fails validation gets REJECTED.json with its errors and a
digest of its files and is skipped until they change; one already
committed waits for --replace.

A changes row names the converter in source and lists the steps tables
the batch wrote in steps_tables, since one batch can span several
(lab, embodiment, end_effector) tables.

commit.py works on local paths and s3:// roots alike (R2 credentials
from the environment). Frame counts come from the MP4 index, so the
committer no longer needs torchcodec. Six concurrent appends on R2 all
land with the expected version count (tests/integration/test_lance_r2).

https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw

@RyanPCo
RyanPCo force-pushed the ryanco/lance-s3-committer branch 2 times, most recently from 5950ffc to f4da635 Compare October 23, 2026 19:28
@RyanPCo
RyanPCo force-pushed the ryanco/lance-mp4-ranges branch from 07c145d to 12ee83c Compare October 23, 2026 19:36
@RyanPCo
RyanPCo force-pushed the ryanco/lance-s3-committer branch from f4da635 to 13c02e0 Compare October 23, 2026 19:36
@RyanPCo
RyanPCo force-pushed the ryanco/lance-mp4-ranges branch from 12ee83c to ffa99f1 Compare October 23, 2026 20:03
@RyanPCo
RyanPCo force-pushed the ryanco/lance-s3-committer branch 2 times, most recently from 9928b36 to 6a929f6 Compare October 30, 2026 18:27

@RyanPCo RyanPCo left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

try:
ok, bad = LOADERS[source]([d])
except Exception as e:
ok, bad = [], {d.name: [f"unreadable: {type(e).__name__}: {e}"]}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium (error-handling): _load turns every exception into an unreadable: rejection, and commit_staged writes it to REJECTED.json with the current files digest, so pending() skips that delivery until its files are re-uploaded. For aria, sql_rows makes a network round trip to Postgres. A DB outage or timeout during one run therefore rejects the whole batch for good, and so does no SQL row when the SQL metadata lands after the delivery. Let infrastructure errors (SQL, boto, network) propagate and abort the run, or mark these notes retryable so pending() returns them next time. Only content validation failures should be sticky.

Comment thread egomimic/scripts/lance_commit.py Outdated
"""LOADERS[source] over `dirs`; a delivery the loader cannot even read is rejected alone."""
try:
return LOADERS[source](dirs)
except Exception:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium (error-handling): The broad except Exception also catches infra failures. If sql_rows fails (DB down, auth), the per-dir retry fails the same way, and every delivery in the batch gets REJECTED.json with its current digest. pending then skips them permanently until someone restages. Catch only per-delivery read/parse errors here (OSError, JSON/Arrow decode, KeyError on meta), or skip writing REJECTED.json for errors that aren't about the delivery, so transient outages fail the run instead.

frames = VideoDecoder(
str(d / f"{cam}.mp4"), seek_mode="exact"
).metadata.num_frames
frames = mp4_ranges.read_index(

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low (robustness): Frame-count validation now goes through mp4_ranges.read_index, whose top-level loop does pos += size without handling size == 0, which means the box extends to EOF. ffmpeg writes mdat with size 0 until it writes the trailer, so an MP4 from a killed encode (no moov) makes this loop spin forever and the committer hangs instead of rejecting the delivery. Have read_index raise ValueError when size is 0 on a non-moov box or smaller than the header, so _load rejects the file.

Comment thread egomimic/rldb/lance/commit.py Outdated
return None


def _open(uri: str, version=None) -> lance.LanceDataset:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low (duplication): _open is the same as dataset.open_table, which already exists: lance.dataset(uri, version=version, storage_options=storage_options(uri)). Import open_table here and drop _open. Then annotations.py, scripts/lance_annotations.py and the migrate script in later PRs can use the public helper instead of importing a private one from commit.

for vid, errs in rejected.items():
note = {
"errors": errs,
"digest": batch[vid],

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low (function-too-long): batch[vid] and local[...] are keyed by staging dir name, but the loaders key rejected by meta.json vendor_recording_id (only the _load fallback uses d.name). A delivery staged under a dir whose name differs from its meta id raises KeyError after commit_batch has committed, and the batch is never unstaged. Key everything by the staging dir, or have the loaders reject a delivery whose meta id doesn't match its dir.

return "/".join([str(root).rstrip("/"), *parts])


def _split(uri: str) -> tuple[str, str]:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit (duplication): _split is the third copy of uri[len("s3://"):].split("/", 1); dataset.py _video_source and read_range already do this inline. Consider putting it in dataset.py next to _is_remote/_s3_client and making those helpers public, since commit.py and the migrate script both import them across modules.

Comment thread egomimic/rldb/lance/commit.py Outdated
for u in uris:
if not _is_remote(u):
Path(u).unlink(missing_ok=True)
for i in range(0, len(remote), 1000):

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit (magic-number): Name the 1000, e.g. _S3_DELETE_MAX = 1000 # DeleteObjects per-request limit, so the reason for the chunking is visible.

from egomimic.rldb.lance.schema import recording_id, steps_path # noqa: E402
from egomimic.scripts.lance_commit import REJECTED, commit_staged # noqa: E402

EVA_STEPS = steps_path("rl2", "eva_bimanual", "parallel_jaw_gripper")

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit (naming): EVA_STEPS is schema.EVA_STEPS (a pa.Schema) elsewhere in the package and fixtures. Here it's a steps-table path. Rename it to EVA_STEPS_PATH to avoid confusion.

The committer (egomimic/scripts/lance_commit.py, moved from
lance_pilot/commit.py) is the only writer of the Lance tables.
Converters stage deliveries under <staging>/<lab>/<vendor id>/ with
meta.json uploaded last (commit.stage; --stage-to on aria_to_lance,
eva_to_lance and the Ray driver, which no longer commits). Each run
validates complete deliveries, commits them in batches, records each
batch in a new changes table and tags every table <lab>-<UTC time>,
then deletes their staged files. A delivery that fails validation gets
REJECTED.json with its errors and a digest of its files and is skipped
until they change; one already committed waits for --replace.

commit.py works on local paths and s3:// roots alike (R2 credentials
from the environment). Frame counts come from the MP4 index, so the
committer no longer needs torchcodec. Six concurrent appends on R2 all
land with the expected version count (tests/integration/test_lance_r2).

Claude-Session: https://claude.ai/code/session_01Kt1E8T2x8kdV9akAvbJyVd
…bles

Staging prefixes, tag prefixes and the committer's --lab named the converter
(aria, eva), which is no longer what lab means in the tables. They are now
--source / <staging>/<source>/. The changes table drops its lab column for
source (the converter or "annotations") and steps_tables, the
steps/{lab}/{embodiment}/{end_effector}.lance tables the batch wrote, since one
batch can span several. Staged deliveries carry lab and end_effector.

Claude-Session: https://claude.ai/code/session_01HjCQBqQCr2igFEMR7yUovw
@RyanPCo

RyanPCo commented Oct 30, 2026

Copy link
Copy Markdown
Contributor Author

Review findings: status after the fixup pass. The fixes are squashed into this PR's commits, and the whole stack was force-pushed.

Fixed

  • Transient failures are written into REJECTED.json, so the delivery is never retried: SQL and connection errors now propagate and abort the run with no notes written. pending() retries notes whose only error is 'no SQL row'; tests added
  • Validation through mp4_ranges.read_index can hang on a truncated MP4: Fixed in read_index (rldb/lance: stream frames from MP4 byte ranges, drop chunk tables #730 fixup). The ValueError is caught by _load's per-delivery fallback, so the delivery is rejected instead of hanging
  • Magic 1000 batch size in remove(): Added _S3_DELETE_MAX; groups keys by bucket first, then chunks within each bucket
  • Broad except Exception retry in _load hides loader bugs: Only file and parse errors (OSError except connection/timeout, ValueError, KeyError, TypeError, IndexError, struct.error) trigger the per-delivery fallback. Others propagate; tests use a sqlalchemy OperationalError and a ConnectionError
  • (dup check) see read_index items: Duplicate of the truncated-MP4 entry above

Partial

  • commit._open duplicates dataset.open_table: commit._open is now an alias of dataset.open_table; scripts/lance_annotations.py uses open_table. annotations.py and the migrate script still import _open
  • commit_staged does download, load, skip, commit, reject-note, unstage and cleanup inline: Fixed the KeyError: _load rejects a staged dir whose meta vendor id differs from the dir name, keyed by the dir name, so every key matches (test added). The skipped-id set was already built once per batch. The function split and listing reuse were not done

Not fixed

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant