From 0850626d481373bd3de8be9221eb64a726e92e6e Mon Sep 17 00:00:00 2001 From: Arpit Gupta Date: Sat, 18 Jul 2026 21:50:39 -0400 Subject: [PATCH] =?UTF-8?q?feat(catalog):=20add=20the=20catalog=20(Phase?= =?UTF-8?q?=201)=20=E2=80=94=20queryable=20episode=20registry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns Forge from a per-dataset tool into a system of record: an append-only set of Parquet tables (episodes, quality_scores) written with pyarrow and queried with embedded DuckDB, on a local directory or an s3:// / gs:// bucket. forge catalog init ./forge-catalog forge ingest ./my_dataset --catalog ./forge-catalog forge query "SELECT task, count(*) FROM episodes GROUP BY task" -c ./forge-catalog forge catalog stats --catalog ./forge-catalog New module forge/catalog/: - schema.py — pyarrow schemas, SCHEMA_VERSION, SCORER_VERSION, validate_rows() - writer.py — the only writer; atomic part-files + manifest-last commit protocol - catalog.py — Catalog class: init/open (version guard), sql(), commit_batch, stats(), views (episodes, quality_scores, v_latest_quality); manifest-driven reads so queries never see a partial batch - ingest.py — discover -> content-hash -> skip-if-exists -> register -> score - cli.py — forge catalog init/stats, forge ingest, forge query Design notes: - Reuses existing internals — the readers behind `forge inspect` for metadata and QualityAnalyzer.analyze_episode (the engine behind `forge quality`) for scoring. No metadata/scoring logic reimplemented; no existing behavior changed. - Idempotent: episodes keyed by an xxh3 content hash of their proprio stream, so re-running an ingest over the same sources is a no-op. - Crash-safe: each flush commits atomically (manifest written last); reads load only manifested part-files. - Cloud-native: catalog and/or source can live in S3/GCS (reuses forge.io fsspec; DuckDB reads via the same fsspec client — works with real S3 and MinIO). - New deps behind the [catalog] extra: duckdb, xxhash, pyarrow. The base CLI never imports them; they load only when a catalog command runs. Docs: README "The catalog" section, forge/catalog/README.md, a runnable droid_100 example (forge/catalog/catalog_example_droid_100/, excluded from the built package), the design RFC (docs/forge_data_engine_design.md), and a CHANGELOG entry. Tests: tests/test_catalog.py (22 tests) — schema validation, commit protocol + crash recovery, idempotent re-ingest, v_latest_quality version selection, synthetic-dataset integration, and memory:// cloud. No existing tests change. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 33 ++ README.md | 39 ++ docs/forge_data_engine_design.md | 287 +++++++++++++ forge/catalog/README.md | 137 ++++++ forge/catalog/__init__.py | 57 +++ forge/catalog/catalog.py | 326 ++++++++++++++ .../catalog_example_droid_100/README.md | 88 ++++ .../_checkpoint.json | 6 + ...fest-61af2f5eebcb4d7ea861b08f99cf5b5b.json | 17 + .../catalog_example_droid_100/catalog.json | 9 + ...t-923c824ae5744b248bb406c3509d0da7.parquet | Bin 0 -> 15821 bytes ...t-c39e4fef7f7240b5b4a385b97692365b.parquet | Bin 0 -> 14484 bytes forge/catalog/cli.py | 244 +++++++++++ forge/catalog/ingest.py | 344 +++++++++++++++ forge/catalog/schema.py | 130 ++++++ forge/catalog/writer.py | 144 +++++++ forge/cli.py | 8 + pyproject.toml | 14 +- tests/test_catalog.py | 398 ++++++++++++++++++ 19 files changed, 2280 insertions(+), 1 deletion(-) create mode 100644 docs/forge_data_engine_design.md create mode 100644 forge/catalog/README.md create mode 100644 forge/catalog/__init__.py create mode 100644 forge/catalog/catalog.py create mode 100644 forge/catalog/catalog_example_droid_100/README.md create mode 100644 forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/_checkpoint.json create mode 100644 forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/manifest-61af2f5eebcb4d7ea861b08f99cf5b5b.json create mode 100644 forge/catalog/catalog_example_droid_100/catalog.json create mode 100644 forge/catalog/catalog_example_droid_100/episodes/ingest_date=2026-07-18/part-923c824ae5744b248bb406c3509d0da7.parquet create mode 100644 forge/catalog/catalog_example_droid_100/quality_scores/ingest_date=2026-07-18/part-c39e4fef7f7240b5b4a385b97692365b.parquet create mode 100644 forge/catalog/cli.py create mode 100644 forge/catalog/ingest.py create mode 100644 forge/catalog/schema.py create mode 100644 forge/catalog/writer.py create mode 100644 tests/test_catalog.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ec67df9..fc8484b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **The catalog (Phase 1) — an append-only, queryable registry of episodes.** + Turns Forge from a per-dataset tool into a system of record: a set of + append-only Parquet tables (`episodes`, `quality_scores`) written with pyarrow + and queried with embedded DuckDB, on a local directory or an `s3://` / `gs://` + bucket. + + ```bash + pip install "forge-robotics[catalog]" + forge catalog init ./forge-catalog + forge ingest ./my_dataset --catalog ./forge-catalog + forge query "SELECT task, count(*) FROM episodes GROUP BY task" -c ./forge-catalog + forge catalog stats --catalog ./forge-catalog + ``` + + - New module `forge/catalog/` with a `Catalog` class as the single entry + point (`from forge.catalog import Catalog`), plus `forge/catalog/ingest.py`. + - New CLI commands: `forge catalog init`, `forge catalog stats`, + `forge ingest`, `forge query` (SQL over `episodes`, `quality_scores`, and a + `v_latest_quality` view; `--format table|json|csv`). + - **Ingestion reuses existing internals** — the same format readers behind + `forge inspect` for metadata and `QualityAnalyzer.analyze_episode` (the + engine behind `forge quality`) for scoring. No metadata extraction or + scoring logic was reimplemented, and no existing behavior changed. + - **Idempotent & crash-safe.** Episodes are keyed by a content hash, so + re-running an ingest over the same sources is a no-op. Each flush commits + atomically via a manifest-last protocol; queries read only manifested + part-files, so a crash never exposes a partial batch. + - New deps behind the `[catalog]` extra: `duckdb`, `xxhash` (and an explicit + `pyarrow`). The base CLI never imports them — they load only when a catalog + command runs. + - See [forge/catalog/README.md](forge/catalog/README.md) and + [docs/forge_data_engine_design.md](docs/forge_data_engine_design.md). + - **Cloud storage support (`s3://`, `gs://`).** Every command that accepts a dataset path now also accepts Amazon S3 and Google Cloud Storage URIs, in addition to local paths and `hf://` URLs. For example: diff --git a/README.md b/README.md index 695e876..cf704fe 100644 --- a/README.md +++ b/README.md @@ -250,6 +250,45 @@ forge dedup ./my_dataset ./deduped --method dhash # phash (default) | dha See [forge/dedup/README.md](forge/dedup/README.md) for the algorithm and tuning. +## The catalog + +Forge is a per-dataset tool by default. The **catalog** turns it into a *system of record*: an append-only set of Parquet tables that registers every episode you ingest and annotates it with quality scores, all queryable with SQL. It's zero-server (just Parquet + embedded DuckDB), works on a local directory or an `s3://` / `gs://` bucket, and is readable by pandas/Polars/Spark without Forge. + +```bash +pip install "forge-robotics[catalog]" + +# 1. Create a catalog (local dir or cloud bucket) +forge catalog init ./forge-catalog + +# 2. Ingest datasets — registers + quality-scores each episode. +# Re-running is a no-op (episodes are skipped by content hash). +forge ingest ./my_dataset --catalog ./forge-catalog +forge ingest s3://lab-bucket/raw/2026-07-18/ -c ./forge-catalog + +# 3. Query with SQL (views: episodes, quality_scores, v_latest_quality) +forge query "SELECT task, count(*) FROM episodes GROUP BY task" -c ./forge-catalog +forge query "SELECT e.language_instruction, q.overall_score + FROM episodes e JOIN v_latest_quality q USING(episode_id) + ORDER BY q.overall_score DESC LIMIT 10" -c ./forge-catalog --format json + +# 4. Summary stats +forge catalog stats --catalog ./forge-catalog +``` + +Python API: + +```python +from forge.catalog import Catalog +from forge.catalog.ingest import ingest + +cat = Catalog.init("s3://lab-bucket/forge-catalog") +ingest(["s3://lab-bucket/raw/2026-07-18/"], cat) +df = cat.sql("SELECT robot, avg(overall_score) FROM episodes " + "JOIN v_latest_quality USING(episode_id) GROUP BY robot").to_pandas() +``` + +Ingestion reuses Forge's existing readers (the metadata behind `forge inspect`) and scorer (the engine behind `forge quality`), so the catalog stays consistent with the rest of the toolkit. Writes go through pyarrow; reads through DuckDB; nothing else touches catalog files. See [forge/catalog/README.md](forge/catalog/README.md) for the architecture, storage layout, and commit protocol. + ## Dataset Registry A curated catalog of 23+ prominent robotics datasets — browse, search, and download by name instead of memorizing URIs. **[Browse the registry online](https://arpitg1304.github.io/forge/registry.html)** diff --git a/docs/forge_data_engine_design.md b/docs/forge_data_engine_design.md new file mode 100644 index 0000000..4456072 --- /dev/null +++ b/docs/forge_data_engine_design.md @@ -0,0 +1,287 @@ +# Forge Data Engine — Design Doc + +**Status:** Draft / RFC +**Scope:** End-to-end lifecycle of robot demonstration data in Forge: ingestion → registration → quality scoring → embedding → dedup → query/curation → snapshots → training-set export. +**Non-goals (for now):** Distributed execution (Ray), a hosted service, per-frame vector search at scale, real-time streaming ingest. + +--- + +## 1. Motivation + +Forge today is a *tool*: it operates on one dataset at a time (`forge convert`, `forge quality`, `forge filter`). Labs that collect data continuously — thousands of episodes per day across operators, robots, and tasks — need a *system of record*: a single growing corpus where every episode is registered once, annotated with derived signals (quality, embeddings, dedup), and queryable for curation. + +This doc defines that system. Design goals: + +1. **Append-only and immutable.** Raw episode data lands once and never changes. All derived facts are versioned annotations, never in-place mutations. +2. **Zero-server.** The store is Parquet files in a directory or object store bucket. The query engine is embedded (DuckDB). No database to operate, no migrations to run against a live service. +3. **Open and portable.** Everything is readable by pandas / Polars / Daft / Spark without Forge. Forge adds convenience, not lock-in. +4. **Reproducible.** Any training set ever produced can be reconstructed exactly from a snapshot ID. +5. **Incremental.** Daily ingest of N new episodes costs O(N) work (plus one ANN pass for dedup), regardless of corpus size. + +## 2. Architecture Overview + +``` + ┌────────────────────────────────────────────┐ + │ FORGE DATA ENGINE │ + │ │ + teleop rigs ──┐ │ ┌─────────┐ ┌──────────────────────┐ │ + sim rollouts ──┼──────►│ │ INGEST │──►│ episodes (registry) │ │ + hf:// / s3://──┘ │ └────┬────┘ └──────────────────────┘ │ + │ │ per-episode, checkpointed │ + │ ▼ │ + │ ┌──────────┐ ┌──────────┐ ┌─────────┐ │ + │ │ quality │ │ embed │ │ dedup │ │ + │ │ _scores │ │ dings │ │ _edges │ │ + │ └──────────┘ └──────────┘ └─────────┘ │ + │ │ │ │ │ + │ └──────┬──────┴─────────────┘ │ + │ ▼ │ + │ ┌────────────┐ ┌──────────────┐ │ + │ │ QUERY (SQL │───►│ snapshots │ │ + │ │ / DuckDB) │ │ + export │ │ + │ └────────────┘ └──────────────┘ │ + └────────────────────────────────────────────┘ +``` + +Two layers: + +- **Data plane:** raw episode files (LeRobot/RLDS/HDF5/…) stored wherever the lab keeps them (`s3://`, `gs://`, local). Forge references them by URI; it does not copy or own them. +- **Catalog plane:** a set of append-only Parquet tables (the "forgeдb", working name `catalog/`) that describe and annotate the corpus. This doc is mostly about the catalog plane. + +## 3. Storage Layout + +``` +s3://lab-bucket/ +├── raw/ # data plane — lab-owned, any format +│ ├── 2026-07-18/run_0413/... +│ └── ... +└── forge-catalog/ # catalog plane — owned by Forge + ├── catalog.json # version, config, table locations + ├── episodes/ + │ └── ingest_date=2026-07-18/part-*.parquet + ├── quality_scores/ + │ └── ingest_date=2026-07-18/part-*.parquet + ├── embeddings/ + │ └── model_id=siglip-so400m@a1b2/ingest_date=.../part-*.parquet + ├── dedup_edges/ + │ └── computed_date=2026-07-18/part-*.parquet + ├── curation_labels/ + │ └── part-*.parquet # small; unpartitioned append log + └── snapshots/ + ├── snap_2026-07-18_pickplace_v1.json + └── ... +``` + +Rules: + +- **Writes are new files only.** A daily ingest appends `part-*.parquet` files under the day's partition. Nothing rewrites existing files. Concurrent readers are always safe. +- **Partitioning** is by `ingest_date` (and `model_id` for embeddings) — matches the write pattern and prunes well for "recent data" queries. +- **Compaction** is an optional maintenance job (`forge catalog compact`) that merges small files within a closed partition. It replaces files atomically per-partition and is the only rewrite in the system. +- `catalog.json` carries a schema version so future migrations are explicit and detectable. + +## 4. Schemas + +Types below are Arrow/Parquet types. `PK` is logical (Parquet has no constraints; uniqueness is enforced at ingest). + +### 4.1 `episodes` — the registry + +One row per ingested episode. Facts only — nothing derived, nothing that changes. + +| column | type | notes | +|---|---|---| +| `episode_id` | `string` (uuid7) | PK. Time-ordered UUID so sorting by ID ≈ ingest order | +| `content_hash` | `string` | xxh3 of raw episode bytes; exact-dup rejection at ingest | +| `source_uri` | `string` | `s3://…/run_0413`, `hf://…` — where raw data lives | +| `source_format` | `string` | `lerobot-v3`, `rlds`, `hdf5`, `rosbag`, … | +| `robot` | `string` | embodiment id, e.g. `franka`, `aloha` | +| `task` | `string` | task id/name; nullable | +| `language_instruction` | `string` | nullable | +| `operator_id` | `string` | who collected it; nullable | +| `scene_id` | `string` | nullable | +| `num_frames` | `int32` | | +| `duration_s` | `float32` | | +| `fps` | `float32` | nominal control/camera rate | +| `cameras` | `list>` | | +| `state_dim` / `action_dim` | `int16` | | +| `ingest_batch_id` | `string` | groups one ingest run | +| `ingested_at` | `timestamp[us, UTC]` | | +| `extra` | `map` | escape hatch for lab-specific metadata | + +### 4.2 `quality_scores` — versioned derived metrics + +One row per `(episode_id, scorer_version)`. Re-scoring the corpus with an improved scorer appends new rows; old rows remain for reproducibility. + +| column | type | notes | +|---|---|---| +| `episode_id` | `string` | FK → episodes | +| `scorer_version` | `string` | e.g. `q-v3`; maps to a config committed in the repo | +| `overall_score` | `float32` | 0–10 | +| `ldlj` | `float32` | smoothness | +| `dead_action_ratio` | `float32` | | +| `gripper_chatter` | `float32` | | +| `static_ratio` | `float32` | | +| `timestamp_jitter` | `float32` | | +| `saturation_ratio` | `float32` | | +| `action_entropy` | `float32` | | +| `path_length_norm` | `float32` | | +| `flags` | `list` | `jerky`, `mostly_static`, … | +| `computed_at` | `timestamp[us, UTC]` | | + +New metrics = new columns (Parquet schemas can evolve additively) or a `metrics: map` overflow column — start with explicit columns for the 8 shipped metrics, add the map when third-party metrics arrive. + +### 4.3 `embeddings` — versioned per model + +One row per `(episode_id, model_id, level, camera)`. + +| column | type | notes | +|---|---|---| +| `episode_id` | `string` | | +| `model_id` | `string` | `siglip-so400m@` — name **and** checkpoint hash | +| `level` | `string` | `episode` \| `instruction` (per-frame: see §9) | +| `camera` | `string` | which view was embedded; null for `instruction` | +| `pooling` | `string` | `mean`, `first-mid-last`, … | +| `vector` | `fixed_size_list[D]` | D fixed per model_id | +| `computed_at` | `timestamp[us, UTC]` | | + +Mixing vectors across `model_id`s in one similarity computation is an error; Forge enforces this in every query path. + +### 4.4 `dedup_edges` — similarity facts, not verdicts + +One row per detected near-duplicate pair. **Edges record similarity; which episode "wins" is decided at curation time by policy** (e.g. keep the higher quality score). Baking verdicts into edges would freeze the policy. + +| column | type | notes | +|---|---|---| +| `episode_a` / `episode_b` | `string` | canonical order: a < b | +| `similarity` | `float32` | cosine | +| `model_id` | `string` | embeddings used | +| `method` | `string` | `exact-hash` \| `ann-cosine` | +| `computed_at` | `timestamp[us, UTC]` | | + +### 4.5 `curation_labels` — human/policy decisions (append log) + +| column | type | notes | +|---|---|---| +| `episode_id` | `string` | | +| `label` | `string` | `approved` \| `rejected` \| `held` | +| `reason` | `string` | free text or rule id | +| `labeled_by` | `string` | user or `policy:` | +| `labeled_at` | `timestamp[us, UTC]` | | + +Latest-row-wins per episode (resolved with a window function at query time). Appending a new label supersedes without deleting history. + +### 4.6 `snapshots` — frozen training sets + +One JSON manifest per snapshot (not Parquet — they're small and self-describing): + +```json +{ + "snapshot_id": "snap_2026-07-18_pickplace_v1", + "created_at": "2026-07-18T21:04:00Z", + "created_by": "arpit", + "query": "SELECT ... WHERE q.overall_score > 6 AND ...", + "scorer_version": "q-v3", + "embed_model_id": "siglip-so400m@a1b2", + "episode_ids": ["...", "..."], + "counts": {"episodes": 4812, "frames": 1_912_330}, + "catalog_schema_version": 1 +} +``` + +The manifest records the query **and** the resolved episode list, so it is reproducible even after the catalog grows. + +## 5. Ingestion Pipeline + +`forge ingest --catalog s3://lab-bucket/forge-catalog [--batch-id ...]` + +Per-episode stages, each idempotent and checkpointed: + +``` +discover → hash → dedup-exact → register → score → embed → dedup-ann → commit +``` + +1. **Discover.** Expand the URI to candidate episodes using existing Forge readers (any supported format). Cloud URIs via fsspec. +2. **Hash.** `content_hash = xxh3(raw bytes)` streamed, no full-episode load. +3. **Exact-dup check.** If `content_hash` exists in `episodes`, skip (log as `already_ingested`). This makes ingest safely re-runnable — re-pointing at yesterday's folder is a no-op. +4. **Register.** Extract metadata (reusing `forge inspect` internals) → row in `episodes`. +5. **Score.** Run the current scorer version → row in `quality_scores`. +6. **Embed.** Sample frames (default ≈1 Hz effective), run the configured model, pool → row(s) in `embeddings`. Heaviest stage; GPU if available. +7. **ANN dedup.** Query new episode vectors against the existing corpus index (see §7); similarities ≥ threshold → rows in `dedup_edges`. +8. **Commit.** Stage rows in a local buffer; flush as new Parquet part-files per table at batch end (or every K episodes). A `_manifest.json` per batch lists written files, enabling cleanup of partial batches. + +**Failure model:** a crashed ingest resumes from its per-episode checkpoint file; already-committed episodes are skipped via the exact-dup check. No partial rows: an episode either fully commits across all tables in its flush, or its staged rows are discarded. + +**Ordering note:** stages 5–7 can also run as separate backfill jobs (`forge score`, `forge embed`, `forge dedup`) decoupled from registration — required anyway for re-scoring the corpus when versions bump. + +## 6. Query & Curation Layer + +**Engine: embedded DuckDB.** It reads partitioned Parquet directly (local or S3), joins across tables, and adds zero operational burden. Forge ships views so users never hand-write the joins: + +- `v_latest_quality` — quality rows for the *current* scorer_version +- `v_curation` — latest label per episode +- `v_dup_losers(threshold, policy)` — episodes that lose a dedup pairing under a policy + +CLI surface: + +```bash +forge query "SELECT task, count(*) FROM episodes GROUP BY task" # raw SQL +forge curate --where "quality > 6 AND task = 'pick_place'" \ + --dedup 0.97 --dedup-policy keep-higher-quality \ + --label approved --reason "training mix v2" +forge search "picks up the red cup" --top 20 # text → vector → ANN +``` + +`forge curate` = query → resolve dedup policy → append `curation_labels` rows. It never touches raw data. + +**Python API** mirrors this: `catalog = forge.Catalog("s3://…"); catalog.sql(...)`, plus a thin dataframe-ish helper for common filters. The SQL escape hatch is the power-user API; we do not build an expression engine. + +## 7. Vector Search & Dedup Mechanics + +- **≤ ~100K episodes:** brute-force cosine in DuckDB (`array_cosine_similarity`) is sub-second on episode-level vectors. Ship this first. +- **Beyond that, or per-frame vectors:** move the `embeddings` table to **Lance** format (columnar like Parquet, built-in IVF/HNSW ANN, embeds like DuckDB). This is a storage-format swap for one table, not an architecture change — a deliberate later upgrade, not a v1 dependency. +- Ingest-time ANN keeps a small in-memory index of corpus vectors (or the Lance index) so daily dedup is O(new × log corpus), not O(new × corpus). +- Dedup threshold default **0.97**, conservative on purpose: teleop corpora contain many *legitimately* similar episodes (same task, same scene). `forge dedup --dry-run --html` renders candidate pairs with thumbnails for human calibration before any labels are applied. + +## 8. Versioning & Reproducibility Rules + +1. **Scorer/model changes append, never overwrite.** `scorer_version` and `model_id` are part of every derived row. +2. **Snapshots pin versions.** A snapshot records which scorer_version/model_id its query filtered on — re-running the query later against the same versions yields the same set. +3. **Raw data is immutable.** Corrections (e.g. wrong task label) are handled by appending to `curation_labels` or a future `episode_annotations` table, not by editing `episodes`. +4. **Schema evolution is additive.** New columns are nullable; `catalog.json` bumps `schema_version`; Forge refuses to write to a catalog with a newer schema than it understands. + +## 9. Phasing + +**Phase 1 — Catalog core.** `episodes` + `quality_scores` tables, `forge ingest` (stages 1–5, 8), `forge query`, DuckDB views. Value on day one: a queryable registry with quality over a growing corpus. + +**Phase 2 — Embeddings + search.** `embeddings` table, `forge embed` (backfill + ingest stage 6), `forge search`. Model deps behind `pip install forge-robotics[embed]`. + +**Phase 3 — Dedup + curation.** `dedup_edges`, `curation_labels`, `forge dedup`, `forge curate` with policies, HTML pair-review. + +**Phase 4 — Snapshots + export.** `forge snapshot create/list/materialize`; `materialize` reuses Forge's converters to write the snapshot as LeRobot-v3/RLDS for training. + +**Phase 5 — Scale-outs (as needed).** Lance for ANN; per-frame embeddings (`level='frame'`, unlocks segment-level search using PELT segments); diversity metrics in `forge stats`; Ray for embed backfills. + +## 10. Open Questions + +- **Catalog concurrency:** single-writer assumption for v1 (one ingest job at a time). Do we need a lightweight lock file (`catalog.lock` with lease) before Phase 3, or is convention enough for lab-scale? +- **Per-frame embeddings storage cost:** at 1 Hz, D=1152, a 60 s episode ≈ 270 KB of vectors — fine; at full frame rate it isn't. Confirm 1 Hz default is acceptable for segment search. +- **Task/robot vocabularies:** free-form strings vs. a controlled vocab table? Free-form for v1; revisit when cross-lab sharing matters. +- **Iceberg/Delta instead of raw Parquet dirs?** They add real transactionality but heavier deps. Current position: raw partitioned Parquet + manifest files is enough at lab scale; the layout is Iceberg-compatible if we ever promote it. + +## Appendix A — Example end-to-end session + +```bash +# Nightly cron +forge ingest s3://lab-bucket/raw/2026-07-18/ --catalog s3://lab-bucket/forge-catalog + +# Researcher, next morning +forge query "SELECT task, avg(overall_score) FROM v_latest_quality \ + JOIN episodes USING (episode_id) GROUP BY task" +forge search "regrasps after a failed pick" --top 20 +forge dedup --dry-run --html review.html +forge curate --where "quality > 6 AND task='pick_place'" --dedup 0.97 \ + --dedup-policy keep-higher-quality --label approved + +# Training time +forge snapshot create pickplace_v2 --where "label = 'approved'" +forge snapshot materialize pickplace_v2 ./train_pickplace_v2 --format lerobot-v3 +``` diff --git a/forge/catalog/README.md b/forge/catalog/README.md new file mode 100644 index 0000000..53c3669 --- /dev/null +++ b/forge/catalog/README.md @@ -0,0 +1,137 @@ +# `forge.catalog` + +An append-only, queryable registry of episodes — the system-of-record layer for +Forge. This is **Phase 1** of the [Forge Data Engine design](../../docs/forge_data_engine_design.md): +the `episodes` and `quality_scores` tables plus ingest, query, and stats. + +## Why + +Forge on its own operates one dataset at a time. Labs that collect thousands of +episodes a day need a single growing corpus where every episode is registered +once, annotated with derived signals (quality now; embeddings/dedup later), and +queryable for curation. The catalog is that corpus. + +Design commitments (Phase 1 slice of them): + +- **Append-only & immutable.** Raw data is never touched; derived facts are + versioned annotations. Every write is a new part-file; nothing is rewritten. +- **Zero-server.** Parquet files in a directory or bucket, queried by embedded + DuckDB. No database to run. +- **Open & portable.** Plain partitioned Parquet — readable by pandas / Polars / + Spark without Forge. +- **Local or cloud, identical.** Local dir, `s3://`, `gs://`, or `memory://` + behave the same (fsspec under the hood). + +## Architecture + +``` + Catalog (single entry point) + / \ + WRITES (pyarrow) READS (DuckDB / SQL) + | | + writer.py | | catalog.sql(), views + (part-files + manifests) (reads only manifested files) + | | + └───── Parquet tables in / ─────┘ +``` + +Strict division of labor: + +- **`schema.py`** — the pyarrow schemas, `SCHEMA_VERSION`, `SCORER_VERSION`, and + `validate_rows()` (every write is validated, failing loudly on mismatch). +- **`writer.py`** — the *only* module that writes into a catalog directory. +- **`catalog.py`** — the `Catalog` class: `init`/`open`, `sql()`, `commit_batch`, + `stats()`, and the DuckDB views. Reads only. +- **`ingest.py`** — the ingest pipeline (discover → hash → skip → register → + score → stage), reusing existing Forge readers and the quality scorer. +- **`cli.py`** — `forge catalog init/stats`, `forge ingest`, `forge query`. + +> **Invariant:** nothing outside `writer.py` writes catalog files; nothing reads +> catalog data except through `Catalog.sql()`. + +## Storage layout + +``` +/ +├── catalog.json # schema_version, forge version, created_at +├── episodes/ingest_date=YYYY-MM-DD/part-.parquet +├── quality_scores/ingest_date=YYYY-MM-DD/part-.parquet +└── _batches// + ├── manifest-.json # commit marker (one per flush) + └── _checkpoint.json # progress marker (mutable) +``` + +## Commit protocol (atomicity) + +1. A flush writes all its part-files first. +2. It then writes a uniquely-named `manifest-.json` **last** — the commit + marker listing exactly the files it produced. +3. **Reads only ever load files named in a committed manifest.** Part-files left + by a crash before its manifest landed are invisible to every query — so a + query never sees a partially-committed batch. + +On local filesystems each file is written to a temp name and renamed into place +(atomic appearance); on object stores the PUT is itself atomic. + +Recovery is automatic: a crashed ingest leaves orphan part-files and no manifest; +re-running skips already-committed episodes by **content hash** and re-does only +the lost ones. Orphan detection is exposed (`Catalog.orphan_part_files()`) so a +future `gc` can clean them — Phase 1 leaves the convention in place but does not +delete. + +## Schemas + +`episodes` — one row per ingested episode, facts only (id, `content_hash`, +source, robot/task/language, frame/time/fps, cameras, dims, ingest provenance, +and an `extra` map escape hatch). + +`quality_scores` — one row per `(episode_id, scorer_version)`: `overall_score` +plus one column per Forge quality metric (`ldlj`, `sparc`, `dead_fraction`, +`gripper_chatter_rate`, `joint_path_length`, `overall_saturation`, +`mean_entropy`, `psd_high_fraction`, `state_action_consistency`, +`static_fraction`, `jitter_ratio`), `flags`, and `computed_at`. Re-scoring the +corpus appends new rows under a new `scorer_version`; old rows remain for +reproducibility. + +`catalog.json` records `schema_version`; Forge refuses to open a catalog written +by a newer version than it understands. + +## Content hashing + +`content_hash` is an xxh3 of the episode's **proprio stream** (state/action/ +timestamp, streamed frame-by-frame) plus a shape signature. It's deterministic +and per-episode, which is what makes re-ingest idempotent. Forge readers expose +logical episodes rather than per-episode byte ranges (multiple episodes often +share one parquet/video file), so hashing content is the portable choice; +images are excluded (expensive to decode, and proprio is a strong fingerprint +for teleop data). + +## Query surface + +Views registered on every connection: + +- `episodes`, `quality_scores` — the raw tables. +- `v_latest_quality` — the quality row for the newest `scorer_version` per + episode (then most recent `computed_at`). + +```python +cat.sql("SELECT task, count(*) FROM episodes GROUP BY task") # -> pyarrow.Table +``` + +## Example + +[`catalog_example_droid_100/`](catalog_example_droid_100/) is a small, ready-to-query +catalog built from the `lerobot/droid_100` dataset (100 episodes, ~44 KB). Its +[README](catalog_example_droid_100/README.md) has example queries and the exact +commands to reproduce it (from a local MinIO `s3://` source, or straight from +`hf://` with no MinIO). + +```bash +forge catalog stats -c forge/catalog/catalog_example_droid_100 +``` + +## Not in Phase 1 + +Embeddings, dedup edges, curation labels, and snapshots (Phases 2–4); +distributed execution, locking, compaction, and gc. The manifest convention is +in place so gc becomes possible later, but it is not implemented. diff --git a/forge/catalog/__init__.py b/forge/catalog/__init__.py new file mode 100644 index 0000000..ea09f1c --- /dev/null +++ b/forge/catalog/__init__.py @@ -0,0 +1,57 @@ +"""The Forge catalog — an append-only, queryable registry of episodes. + +A catalog is a set of Parquet tables (``episodes``, ``quality_scores``) in a +local directory or cloud bucket, written with pyarrow and queried with DuckDB. +It turns Forge from a per-dataset tool into a system of record: every episode a +lab ingests is registered once and annotated with quality, then queryable via +SQL. + +Entry point: + + from forge.catalog import Catalog + from forge.catalog.ingest import ingest + + cat = Catalog.init("s3://lab-bucket/forge-catalog") + ingest(["s3://lab-bucket/raw/2026-07-18/"], cat) + cat.sql("SELECT task, count(*) FROM episodes GROUP BY task") + +Package-level imports are lazy (PEP 562) so merely importing :mod:`forge` — or +running ``forge --help`` on a base install — never pulls in the optional +``pyarrow`` / ``duckdb`` dependencies; they load only when a catalog symbol is +used. (The ``ingest`` pipeline lives at :mod:`forge.catalog.ingest` to avoid a +name clash between the submodule and the function.) +""" + +from __future__ import annotations + +import importlib +from typing import TYPE_CHECKING + +__all__ = [ + "Catalog", + "CatalogError", + "CatalogSchemaError", + "SCHEMA_VERSION", + "SCORER_VERSION", +] + +# name -> submodule that defines it. importlib.import_module bypasses this +# package's __getattr__, so there's no re-entrancy. +_LAZY = { + "Catalog": "forge.catalog.catalog", + "CatalogError": "forge.catalog.catalog", + "CatalogSchemaError": "forge.catalog.schema", + "SCHEMA_VERSION": "forge.catalog.schema", + "SCORER_VERSION": "forge.catalog.schema", +} + +if TYPE_CHECKING: + from forge.catalog.catalog import Catalog, CatalogError + from forge.catalog.schema import SCHEMA_VERSION, SCORER_VERSION, CatalogSchemaError + + +def __getattr__(name: str): + module = _LAZY.get(name) + if module is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + return getattr(importlib.import_module(module), name) diff --git a/forge/catalog/catalog.py b/forge/catalog/catalog.py new file mode 100644 index 0000000..5087d12 --- /dev/null +++ b/forge/catalog/catalog.py @@ -0,0 +1,326 @@ +"""The Catalog — the single entry point for a Forge catalog. + +Division of labor (enforced by convention): + +* **Writes** go exclusively through :class:`~forge.catalog.writer.CatalogWriter` + (pyarrow → Parquet part-files + manifests). +* **Reads** go exclusively through :meth:`Catalog.sql` (DuckDB over the Parquet + tables). DuckDB never writes to the catalog. + +DuckDB reads the catalog's Parquet through the *same* fsspec filesystem the +writer uses (registered via ``register_filesystem``), so a catalog on local +disk, ``s3://``, ``gs://``, or ``memory://`` is queried identically — including +against a local MinIO, since the S3 credentials/endpoint come from the shared +fsspec client rather than DuckDB's own httpfs config. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any + +from forge.catalog.schema import ( + SCHEMA_VERSION, + TABLES, + CatalogSchemaError, + validate_rows, +) +from forge.catalog.writer import CatalogWriter, _normalize_protocol +from forge.core.exceptions import ForgeError +from forge.io.paths import get_filesystem + +if TYPE_CHECKING: + import pyarrow as pa + + +def _forge_version() -> str: + try: + from importlib.metadata import version + + return version("forge-robotics") + except Exception: + return "unknown" + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +class CatalogError(ForgeError): + """Raised for catalog-level problems (missing, incompatible version, …).""" + + +class Catalog: + """A Forge catalog rooted at a local path or cloud URI.""" + + def __init__(self, root: str): + self.root_uri = str(root) + self._fs, self._root = get_filesystem(self.root_uri) + self._is_local = bool( + _normalize_protocol(self._fs.protocol) & {"file", "local"} + ) + self._writer = CatalogWriter(self.root_uri) + self._meta = self._load_catalog_json() + + # -- lifecycle ---------------------------------------------------------- + + @classmethod + def init(cls, root: str, *, exist_ok: bool = False) -> Catalog: + """Create a new catalog at ``root`` (writes ``catalog.json``).""" + fs, base = get_filesystem(str(root)) + catalog_json = "/".join([base.rstrip("/"), "catalog.json"]) + if fs.exists(catalog_json) and not exist_ok: + raise CatalogError( + f"Catalog already exists at {root} (catalog.json present). " + "Pass exist_ok=True to reuse it." + ) + if not fs.exists(catalog_json): + writer = CatalogWriter(str(root)) + writer.write_catalog_json( + { + "schema_version": SCHEMA_VERSION, + "forge_version": _forge_version(), + "created_at": _utc_now().isoformat(), + "tables": sorted(TABLES), + } + ) + return cls(str(root)) + + @classmethod + def open(cls, root: str) -> Catalog: + """Open an existing catalog, refusing a newer schema version.""" + return cls(str(root)) + + def _catalog_json_path(self) -> str: + return "/".join([self._root.rstrip("/"), "catalog.json"]) + + def _load_catalog_json(self) -> dict: + path = self._catalog_json_path() + if not self._fs.exists(path): + raise CatalogError( + f"No catalog at {self.root_uri} (catalog.json not found). " + "Run `forge catalog init` first." + ) + import json + + with self._fs.open(path, "rb") as f: + meta = json.loads(f.read().decode("utf-8")) + version = int(meta.get("schema_version", 0)) + if version > SCHEMA_VERSION: + raise CatalogError( + f"Catalog at {self.root_uri} has schema_version {version}, but " + f"this Forge understands up to {SCHEMA_VERSION}. Upgrade Forge." + ) + return meta + + @property + def schema_version(self) -> int: + return int(self._meta.get("schema_version", SCHEMA_VERSION)) + + # -- writes ------------------------------------------------------------- + + def commit_batch( + self, + *, + episodes: list[dict] | None = None, + quality_scores: list[dict] | None = None, + batch_id: str | None = None, + ingest_date: str | None = None, + ) -> str: + """Validate and atomically commit a batch across tables. Returns batch_id. + + Episodes and their quality scores commit together (manifest written + last), so an episode is either fully present across both tables or not + at all. + """ + import uuid + + batch_id = batch_id or uuid.uuid4().hex + ingest_date = ingest_date or _utc_now().date().isoformat() + + tables: dict[str, pa.Table] = {} + if episodes: + schema, _ = TABLES["episodes"] + tables["episodes"] = validate_rows(episodes, schema, table="episodes") + if quality_scores: + schema, _ = TABLES["quality_scores"] + tables["quality_scores"] = validate_rows( + quality_scores, schema, table="quality_scores" + ) + if tables: + self._writer.commit_batch(batch_id, tables, ingest_date=ingest_date) + return batch_id + + def register_episodes(self, rows: list[dict], *, batch_id: str | None = None) -> str: + """Commit a batch of ``episodes`` rows.""" + return self.commit_batch(episodes=rows, batch_id=batch_id) + + def add_quality_scores( + self, rows: list[dict], *, batch_id: str | None = None + ) -> str: + """Commit a batch of ``quality_scores`` rows (e.g. a re-score backfill).""" + return self.commit_batch(quality_scores=rows, batch_id=batch_id) + + # -- reads -------------------------------------------------------------- + + def _read_base(self) -> str: + # DuckDB reads local files by path; cloud/memory via a registered fs by + # URI (scheme preserved so it routes to the right filesystem). + return self._root if self._is_local else self.root_uri + + def _manifests(self) -> list[dict]: + """Load every committed batch manifest under ``_batches/``.""" + import json + + pattern = "/".join([self._root.rstrip("/"), "_batches", "*", "manifest-*.json"]) + try: + paths = self._fs.glob(pattern) + except FileNotFoundError: + return [] + out: list[dict] = [] + for m in paths: + try: + with self._fs.open(m, "rb") as fh: + out.append(json.loads(fh.read().decode("utf-8"))) + except (OSError, ValueError): + continue + return out + + def _committed_files(self, table: str) -> list[str]: + """Read-URIs of the part-files a manifest has committed for ``table``. + + This is the heart of atomicity: only files listed in a committed + ``manifest-*.json`` are ever read, so part-files left by a crash before + its manifest landed are invisible to every query. + """ + base = self._read_base().rstrip("/") + files: list[str] = [] + for manifest in self._manifests(): + for entry in manifest.get("files", []): + if entry.get("table") == table and entry.get("path"): + files.append(f"{base}/{entry['path']}") + return files + + def orphan_part_files(self) -> list[str]: + """Part-files on disk not referenced by any manifest (crash residue). + + The catalog never reads or deletes these — the manifest convention just + makes a future ``gc`` possible. Exposed mainly for tests. + """ + committed = set() + for manifest in self._manifests(): + for entry in manifest.get("files", []): + if entry.get("path"): + committed.add(entry["path"]) + base = self._root.rstrip("/") + orphans: list[str] = [] + for table in TABLES: + pattern = "/".join([base, table, "*", "part-*.parquet"]) + try: + found = self._fs.glob(pattern) + except FileNotFoundError: + found = [] + for p in found: + rel = p[len(base) + 1:] if p.startswith(base) else p + if rel not in committed: + orphans.append(rel) + return orphans + + def _connect(self): + import duckdb + + con = duckdb.connect() + if not self._is_local: + # Route cloud/memory reads through the same fsspec client the writer + # uses (shared creds/endpoint — works with real S3 and MinIO alike). + con.register_filesystem(self._fs) + + for table, (schema, _partition) in TABLES.items(): + files = self._committed_files(table) + if files: + literal = ", ".join("'" + f.replace("'", "''") + "'" for f in files) + con.execute( + f"CREATE VIEW {table} AS " + f"SELECT * FROM read_parquet([{literal}], union_by_name=true)" + ) + else: + # Empty catalog table — expose the columns so queries still work. + con.register(f"_{table}_empty", schema.empty_table()) + con.execute(f"CREATE VIEW {table} AS SELECT * FROM _{table}_empty") + + # Latest quality row per episode (max scorer_version, then most recent). + con.execute( + """ + CREATE VIEW v_latest_quality AS + SELECT * FROM quality_scores + QUALIFY row_number() OVER ( + PARTITION BY episode_id + ORDER BY scorer_version DESC, computed_at DESC + ) = 1 + """ + ) + return con + + def sql(self, query: str) -> pa.Table: + """Run a read-only SQL query and return a ``pyarrow.Table``. + + Views available: ``episodes``, ``quality_scores``, ``v_latest_quality``. + Call ``.to_pandas()`` on the result for a DataFrame. + """ + con = self._connect() + try: + result = con.execute(query) + # Fully materialize before closing the connection (unlike .arrow(), + # which can return a lazy reader). to_arrow_table() is the current + # name; fetch_arrow_table() is the pre-1.3 spelling. + to_arrow = getattr(result, "to_arrow_table", None) or result.fetch_arrow_table + return to_arrow() + finally: + con.close() + + # -- convenience -------------------------------------------------------- + + def episode_hashes(self) -> set[str]: + """Return the set of ``content_hash`` values already in the catalog. + + Used by ingest for exact-duplicate skipping. Cheap on an empty or small + catalog; reads only the ``content_hash`` column. + """ + tbl = self.sql("SELECT content_hash FROM episodes") + return set(tbl.column("content_hash").to_pylist()) + + def stats(self) -> dict[str, Any]: + """Return summary statistics as canned SQL through :meth:`sql`.""" + totals = self.sql( + "SELECT count(*) AS episodes, " + "coalesce(sum(num_frames), 0) AS frames, " + "coalesce(sum(duration_s), 0.0) AS seconds " + "FROM episodes" + ).to_pylist()[0] + + per_task = self.sql( + "SELECT coalesce(task, '(none)') AS task, count(*) AS n " + "FROM episodes GROUP BY 1 ORDER BY n DESC" + ).to_pylist() + per_robot = self.sql( + "SELECT coalesce(robot, '(none)') AS robot, count(*) AS n " + "FROM episodes GROUP BY 1 ORDER BY n DESC" + ).to_pylist() + + score = self.sql( + "SELECT count(*) AS scored, min(overall_score) AS min, " + "avg(overall_score) AS mean, median(overall_score) AS median, " + "max(overall_score) AS max FROM v_latest_quality" + ).to_pylist()[0] + + return { + "episodes": int(totals["episodes"]), + "total_frames": int(totals["frames"]), + "total_hours": round(float(totals["seconds"]) / 3600.0, 3), + "per_task": per_task, + "per_robot": per_robot, + "overall_score": score, + } + + +__all__ = ["Catalog", "CatalogError", "CatalogSchemaError"] diff --git a/forge/catalog/catalog_example_droid_100/README.md b/forge/catalog/catalog_example_droid_100/README.md new file mode 100644 index 0000000..f41520b --- /dev/null +++ b/forge/catalog/catalog_example_droid_100/README.md @@ -0,0 +1,88 @@ +# Example catalog — `droid_100` + +A small, checked-in example catalog built by ingesting the +[`lerobot/droid_100`](https://huggingface.co/datasets/lerobot/droid_100) dataset +(100 episodes) into the Forge catalog. It shows exactly what a catalog looks like +on disk and gives you real data to run `forge query` / `forge catalog stats` +against without ingesting anything yourself. + +See [`../README.md`](../README.md) for the catalog architecture and +[the design doc](../../../docs/forge_data_engine_design.md) for the bigger +picture. + +## What's in here + +``` +catalog_example_droid_100/ +├── catalog.json # schema version, forge version, created_at +├── episodes/ingest_date=2026-07-18/part-*.parquet # 100 rows — one per episode (facts) +├── quality_scores/ingest_date=2026-07-18/part-*.parquet # 100 rows — quality per episode +└── _batches// + ├── manifest-*.json # commit marker + └── _checkpoint.json # ingest progress marker +``` + +The whole thing is **~44 KB** — the catalog stores derived *facts* (metadata + +quality scores), never the raw ~460 MB of videos. Episodes reference their raw +data by `source_uri`. + +Summary of this catalog: + +| episodes | frames | hours | fps | avg quality | range | +|---:|---:|---:|---:|---:|---:| +| 100 | 32,212 | 0.60 | 15 | 8.26 | 7.40 – 8.96 | + +## Query it (no ingest needed) + +```bash +forge catalog stats -c forge/catalog/catalog_example_droid_100 + +# Top-quality episodes and their language instructions +forge query "SELECT e.language_instruction, round(q.overall_score, 2) AS score + FROM episodes e JOIN v_latest_quality q USING(episode_id) + ORDER BY score DESC LIMIT 10" \ + -c forge/catalog/catalog_example_droid_100 --format json + +# Frame-count distribution +forge query "SELECT min(num_frames), max(num_frames), round(avg(num_frames)) AS avg + FROM episodes" -c forge/catalog/catalog_example_droid_100 +``` + +Because it's plain Parquet, you can also read it with anything else — no Forge: + +```bash +duckdb -c "SELECT source_format, count(*) FROM + 'forge/catalog/catalog_example_droid_100/episodes/*/*.parquet' GROUP BY 1" +# or: python -c "import pandas as pd, glob; +# print(pd.read_parquet(glob.glob('forge/catalog/catalog_example_droid_100/episodes/*/*.parquet')))" +``` + +## How this example was built + +The source `droid_100` was stored in a local MinIO bucket (S3-compatible), so the +ingest read from `s3://` and wrote the catalog locally: + +```bash +# Point s3fs/forge at the local MinIO where droid_100 lives (see forge/io/TESTING_MINIO.md) +export AWS_ACCESS_KEY_ID=minioadmin AWS_SECRET_ACCESS_KEY=minioadmin +export AWS_ENDPOINT_URL=http://127.0.0.1:9000 + +forge catalog init forge/catalog/catalog_example_droid_100 +forge ingest s3://forge-datasets/droid_100 \ + --catalog forge/catalog/catalog_example_droid_100 +``` + +### Reproduce it yourself (no MinIO required) + +`droid_100` is public on the Hugging Face Hub, so anyone can rebuild an identical +catalog straight from `hf://`: + +```bash +pip install "forge-robotics[catalog,lerobot,hub]" + +forge catalog init /tmp/droid_100_catalog +forge ingest hf://lerobot/droid_100 --catalog /tmp/droid_100_catalog +forge catalog stats -c /tmp/droid_100_catalog +``` + +Re-running the ingest is a no-op — episodes are skipped by content hash. diff --git a/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/_checkpoint.json b/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/_checkpoint.json new file mode 100644 index 0000000..5b536ed --- /dev/null +++ b/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/_checkpoint.json @@ -0,0 +1,6 @@ +{ + "batch_id": "96151af7756a4f8fa2e7a7b789e0baaf", + "frames": 32212, + "ingested": 100, + "updated_at": "2026-07-18T23:17:47.464221+00:00" +} \ No newline at end of file diff --git a/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/manifest-61af2f5eebcb4d7ea861b08f99cf5b5b.json b/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/manifest-61af2f5eebcb4d7ea861b08f99cf5b5b.json new file mode 100644 index 0000000..84a757d --- /dev/null +++ b/forge/catalog/catalog_example_droid_100/_batches/96151af7756a4f8fa2e7a7b789e0baaf/manifest-61af2f5eebcb4d7ea861b08f99cf5b5b.json @@ -0,0 +1,17 @@ +{ + "batch_id": "96151af7756a4f8fa2e7a7b789e0baaf", + "files": [ + { + "path": "episodes/ingest_date=2026-07-18/part-923c824ae5744b248bb406c3509d0da7.parquet", + "rows": 100, + "table": "episodes" + }, + { + "path": "quality_scores/ingest_date=2026-07-18/part-c39e4fef7f7240b5b4a385b97692365b.parquet", + "rows": 100, + "table": "quality_scores" + } + ], + "ingest_date": "2026-07-18", + "schema_version": 1 +} \ No newline at end of file diff --git a/forge/catalog/catalog_example_droid_100/catalog.json b/forge/catalog/catalog_example_droid_100/catalog.json new file mode 100644 index 0000000..12f8d73 --- /dev/null +++ b/forge/catalog/catalog_example_droid_100/catalog.json @@ -0,0 +1,9 @@ +{ + "created_at": "2026-07-18T23:17:30.208440+00:00", + "forge_version": "0.1.0", + "schema_version": 1, + "tables": [ + "episodes", + "quality_scores" + ] +} \ No newline at end of file diff --git a/forge/catalog/catalog_example_droid_100/episodes/ingest_date=2026-07-18/part-923c824ae5744b248bb406c3509d0da7.parquet b/forge/catalog/catalog_example_droid_100/episodes/ingest_date=2026-07-18/part-923c824ae5744b248bb406c3509d0da7.parquet new file mode 100644 index 0000000000000000000000000000000000000000..11ffda66abb41c5006920472a3f9571420fcf74c GIT binary patch literal 15821 zcmc(G3wRtym2S7Bc1E_WNT5+hL=GckO{7GT`~3(asAb8PjqONI{8F6Q)m7Ej#3Rix zGqT2B6BKUDb%J6PutQ)2>%7>!*fn2F2#>G{3&9EG;_%1@matCt!a|k>bC>0YyWz5b z^+=ZdAc?=_+x3^O?yjynbiAJPF14qALF&+zT$JI^3blY@Y(+#-69gkQpS1sbvXiQ)* zM+-g9P2JQTmjp&Y4C-o*ZfQhS#h8Gqx@TCP88X8(SrD+muyoy|fo*X|cf=S%4B~2` z%`Mk5oq!QTV+MCE-3plNSP>o9<(g{*x~Hn#Bq}E+$8%saGYEJ<0{MkffKEzul~ZDv zhQWyCg@)sWT(?55smcp`sOQ>hsB#tZnug1R!1FZACa&rQoDo;<-(x$}^;BZpj*2A$ zN*u$`Oxw^g+BB&lFWCcmm|@ta6B^ElCbtbI@SrEcOaoGc1~pxWaf_IqXG((rm{WT7!ru?AH>#*Hw(DKDL!-k%m z&~z*ZvS=zc)v`35I;N>{50=40&2}|nlfa@DhIl%KI-z6=r7_bbs%4sHh?_-iLTn5% znc*^W~2_GOoN8i6&KHXd}_RWf@#Ez}*u;;O?f6ZY#- zH-K5WA?11y>ZUwZa@0_>U}gqU1JhzIELNvBr=b=whdIh?Ti_5Xb4`yD*E6YZ5yycD zw%c-Pj6t?UiCCed1rWla+;lDOQD$1gd@KugXSXJ|u|Oj*5(2x|OkxGD6_U`v$Cevn z*lT&JNL&Y;=q}@-X~LL^#R#`m({xQ@n(c?N_t+$k9Wh~bJmfG56(l@|7$T;f~aO_utWJs$>g46!Xs?m2}~P4XEOpKoa#mh4JofBE$TAIaJVBs(`#r} zpo1^Z0aQ?E!E8NT;)Y4Frs---H{r2FC&D{0(1QWH{96){-^DbYm^S=K_jJ$GH3q9! z9V4I~Yds8GhUys$U|2vA*^j}v$#V^c3b!(p{ymgKcXr^wk-152*9#n`YBsek6^_LS z&_S~dM>Q-**P(L*(Bj!}V`4H#hX#Yz-lDEKHUTscqPo}#mGBVusX5S+=DF#vZlanl z16_j>+XPy={+cC*YB3vNK&{qfNymC-7{K0GU=mFwZV=jpg+{1zThscQOB~o{xllD@ z8hnTbs!CK%hk9-5!fU!p@NCX?frxNYW*IIv+6&MInr($)>-S2YYS=anLm+}-!r47w zWT0~cmW|!jl-EiO$fOyt8O;Xf>TobUBsvdN7dFi;-nEA~I?Q$9AKBKb9k(6nJNLZU|Z!DEE^V~?wu$x73-Q7;}#YjI&h* z5T9ltJ$if+6Xo_oI1cd9-P>NEr;?H99{*~ znt>D89-JX?VFqwoo!NV*`nrp*XCM%$Ku8_V1YwB{N1=e#KzE&VSINN>_T2>3!^+{A zj_ayW7Yz(o^VHVK5>S__j!l6V9ze@-Jp$Nus0sLDY6y*Dzo@RkjRTmXX{#=RgQ`(R zsU{*|>twI)IKU@nA(~=K92x{S{(BN+pg45%o8XQCM-)^M#t6 zngJJ{N_pt$41(A}0HHO2j71Gw{+}($5`$xF@JNa%W;zT}m4_~H6qB&a_R5F1K|+UG z8ejsE)Cv)UU?;X6I`>sdRPIk|a9ho_9lf?P)+=DmwcsEj3pkc_*xq}aLq;tEMc(*A zuc^`y7J;2Yy1?wf3)6`bha;+pun0+TMGgK17g9Nb9GoyPm8mvQ(_D|jmpEw|?s7=z zLYHc2Im`&nuywN8M8I@HRg(|&n?OL%08&^~M-+t_c*>EI1Ev9{x~7^4JwVdX)m;~u zX(PmvaPPfWW8WMT5y#>d01S3$*p^FQiq~Nm&7gI%2t6|z0LH@*C>;W3bTweWe&~HAB_co0`0gDV0sfXND}p)(611|diT!iI<-@Cw5uFgr5@HiHR6)#1KcK#jic63FTS z_E>dw4lNiSvZnwGnMCA5kVG|U5zAouuH`W(l;{Qz;A5JL{h$aAup|>U>g&K6*RHB=))4;z=&E1S;DHg1-1oDRhx~%VEo_Tf$nJEY0TvCA1i;c=5lIYg zz|24f0SnludafG)Kd}2SBGX*CFA?g@v2?@VDw%iPDXA^HlK}#!MJSx_L^JG!n(&&I z>-ugn9ot0qWFy)EW8J>rlz^0$9fpihfK@;MSx|fkJBC0ufOgvevPct&9)};|Gi1|4 z&3ozDI?Ok;_TeI;hh>`#y8|eqs_E$#;x1>Dm#1YGFH-n->3Z3_}emo3ZEK0<=#&I67P)2mlbsRAAr^;Tliu z(Shljp@Jp&5K>fRu5bi|MF#Nnj6gnup%_B~B&3X)y!Q9i9f)v1ZrfvOKvzDTL@Hnb z$UV&Dl4}#)CJu#{ zBbVez(U4!OHbV#EzSGBINOo%3{5sV{W|L%q-4(0kg-uB_6_vW9@T>clw?k}$T z%+O!I@bJ!W{>$#aIexfQ1Nx!}e0%(+=95L>vGZ5gpy#P)@A4bHcO|f z79!bpZID7>!ZzCo0az7g15${~o@JnRM3&-l7aYLlk(3(-p;ibWRj?P14a}oR+K}Wj zO*zsP!d94L1Ih@c$o1s`7nKHl!$YuCEqJps*`*qRp&FSJfsm8 z=84Qg_zY5Qk)0@A9WJNH8UeG&!sTkY4aFmz4k|W)E=+6R)8#@BFhd*30)g9javSE@ zmJXvwAp;Y3loMU9Lu@2C2rz(LPj@f~rIrD$qJ}YOh_rW(AvEeDd_jB@ zn}f<0>8nk`P**jzy3Iv)uY1Vu2of9Mm~uj;4vsc#1F4JQdJ89EJ))2?JZ@pr5jJ6n zmWzZrK(2)90GV5#3U_fis zF@Ui+WC)S@pu}fTIlvwMh%S_7RQIsq8XQ4Ij>!Qp2=!PfK2RNZ07Hw~g=e=Gu##90MJ!i16(E~U0kA%G)l05u>9 zGI$k63&kQoK`K)4z-|*N<%r3 zz#em#I^ z2MHv^t8PHnimEGQNOce=T=*$Nkiy{`4*D1n$83EJ=__^)8JdkUm8cI)it`e~aS)}T zBa{hyzlOXJNf*RM-4JkX<^5e)1!oZ!if5P>vW~L{^t#jhg(T`u+Xv^rDq#1-m#>aj zL`SwQ6N?b63&!P3mxkH=c77QnMN;6!!cvycrrB0xau6pLuQ@e4S@gOy60iR`60@pC zJ}Qgsp*fzMVW!rc4`wnvpAE9b%XSJyp0V=$+0ThN&!^ICjK6J$=7P5KKyz$#$7ptZ z6a?|MKeD0)`2Dr7#J%ii^DuQLh?I!KA9d>YHM=YqP1vI8zrM`S27^^WVSllk>O!6 zSx7UU%!Z2A5vKD6ybtqCm|Ey{Wk-2(d!F;rWP#9OUaa5JhF8qaj%PaCXLdUS`_@-^9=)g4O;!z7z`Gwklxn0Rf^@vX#F*be0uwEu~YB7NUGbO zA%#MHa!G;HEXL=v5k-BPWeD{Ce4yzK0Bf;RCZQRRgD!F z?TQ7DA;^+EyQt+;v9bK9yiLjXt&lP;L36S&o-WeicP!f3P)rvJ zW89A^9r6uHNSF-Lqn+OYXg)~OlRd$rBT@@CEJ%vEsOlB zB=xV*z5>y}9=Hp#rRGJgbwM5&TfZb$%o6#hQpciX^|duYHoK#5Lsd36xU>2x`AVsi zG*zw1uKB9E6V94R!y;%VTWHr?Zc2=#^Z9JPZP8HsKa7;>;7G;H`v2SdwU(Nd|C$`n zf~5>fvxWN`FrheH=&Y=)|Fl%9A85HzG8T=rUnCXk-@fu*>lZZ?!nZyxf4j4#LcdqK z+>*dVsY`oRemPboIe9C&IR^DAF(S(s))jKu?5M`ciUcnti+Qq(Jl|PYSx2)YxiNox z%}92vp!ZhZ`ZvjuG=)Kso%}{?@Tr3@AKX|&$8yKy-@osk_Qlo5c3qy8>Iys3quX2h z*Z#W{!+5g2V<689g(OJ^W5pt=OgGi+;JX5Hv-EcPw<9k;a8E4AWcl;*;rlM}KU4pX zTP5gx*XEjr@objy(Uu=e$(A2~ws`9m%8HJ{2-bk{b?DIhgs}TWWG+nao{> zSNgYZt6{vbLk`Bg>ce%p%vfQ#eu@87-7eVH2)Xvj@%E>xNz*pIv@SW!Nv_kA+V6=a zxAULJ>VnJ|zfA9zT7Ez_`LUM%T|bgb%0ShH(u;Rqc04vRw*AI~HRITdmxkj{Nc72t zPk&yytxgnJ%WkY{H4`jP#(BP{<*K{dNh8r3MzgzjXZRK0loqbI^;Zo*8~E>&|F+PU zdRuUB!k%|X*_Df1U#XoJvo5L$$rx?9rM;zw^3mdxk5qS}yYAiKZiU>E zdL$NqMp;|^l$5DmAKUd<1I=c#`3!lB*vtQ+T_E?~Wj(O|;J;NBivEh0-&U`v0lOnw zJysc$KeEQ$J5OGY=PW;f5iQyHZ7p(Bf9kz4C3y0RF^9ijDs(=ibZ<`sD}gQ~m&>L{ zi<|+S|4PPxbMaYf+clEi2{?tLlz}4~;Yx%(5hh3Y6>Sa>Ch8F&+##Muu~B8h+aa90?Nii(*b4Ph26=TUeeI_-sH{e8u|ts7JSz|zhUK7 zgAIND`o6(U8wce$>$@@j@kirZHa1*`?qA2Zwe9coZ{NGQ;p>~_i#n8{dtVu9$oz97 z^o#&{uA%8%$G@P&UshJfA1(jJ`-_U-?2tc=m4AtAKdxtSeIM6GT%W^rb%iA5aUVuZ zuR}9Zz!TljzkK@{c>s4HrR7p}NXpG}|6l3<|`bRvcaeoT; zZ{uphb0w~JT#w-T4*Fig^)Tl7(DPz*|0wR0xGu!=cX7WBcc75;814hO_Tu_WyuXZV z8J-v7UVuFN@O%xPubC%FAHaPBu6JRr_v8LWTnF*|KHNWvYY=07yl=z2Kj0brd;0NE zqDl}-R0;BpQrn!j$khJWkxyO61*?}o@=~$u?F0Sh;)AO%zIxT?g3oPPx@on2&-dp& zdZGNW@BT~aUH@=?qM`QYzJILuno82{QWNl*A;_!W%#JHNM=`StaK z{`&Qgqi?5w^FG&q_FI#fy9)0%=B~t;uXXqK-sCS{yuat!Z#DG%_{HCM8^+(F|9j}m zW6l#;=PJzGg0&B#?-tDclrKp?!?g`pPSEOEy!blIeZ2eTeTOlp4gJSJ<4&ygAl9kI za}hK;@cc;6m%sE*@YK@tt6%;v%$3aeuo%74B;BodX;_+lja2i5g6kj|S%#X^j!SmfRoJ!NTvjE{~ z(=h43eby-|5_JhwbZSA8s1n`fR;Xz9EmG0xWyznMV1In|%qA|GOoA2s&Vs)o!Ebl? zGZy?zE`H{9h=X6kaX1?g5}vWiHqNf!Xj0-0N1SR9Pt?TkjIEyeGowTEmi&GFEib*Y zVV;yWs$*4+18Ckw(OmOqXnyliG+XXL^Sj5;to?O$EOuqZhm`)Rcemkv!5TDw*oo%1 zQ)q5@5Y6Myq507-FmR}HQt7Vp7UTWbE}Fksi{?>==Gg+8$q%7<@zWT1L;bB7c*TeD z{?kKfKJ;}odj*N7=j~V8s&2dl@9*%@?AVNEehE$eK{UTVf+jqU=Jr=HdPV(*!P)LX zye9>njem{iv14d<{cn8UJAX=1t3G%s-j55$iv-O-?MBm#rDO58+=kEBinZFrmt)VM zNxz8Z6Tijhcd8%3=LJGPf4?2g{}B_9--_m~A4hY+PtaWP|1t3K3qJ~JR_S=0*Y{DS zuIh+T=W~0|d;`)}E&e>3AA#vub>#t&e7YU)@4o`g%^T5Z88nN8@}K*2G#A{9f$y(9 z2$BzJcz@3(H2q*8_WXNpSCUn$w&GnKLDMLRulxd#bL#S9PjLiPgTzTEL4BS#6<0> z(;|_&8{P(b#)H?%QgkXYRVmBiVgZt$YK|X(D?L-r=004jiiKy`P!%~L0M+PJdvIc* z^?+YpG=^wQ(O9CfMdOIZ6^(ZV8m;HZIX1eOaZZgPW}H)FiW%qBSYpOGHMW>>PK_gG zoKxe98Ryh^V#b?nsyy*47NYjn!4`ooLC$k`*^kEIu4lD{h^5$B9^X#cdNy)arqXgA=W}>TMMtpGY09&Q*MJ z;^Gt46BTz&w5ju_D(;(T&&@wn@xa6-hvy%z_{_wmC+0s<@!5$Eb-}TU&rMvGTX4MM z3lp7(7o4bgd}8T|1ybdc6RKKMSNZh`EmxDQe0oAZT%%S#J7Jut>8^Zk!c-RyRDOTL z$}QYh`GX1j@WNc>ixbX?g%g!8P2l%j_*GTYYt5FFm{-6-+6aj+k>WM2Qeys$)M?f# zt!{>JQFo1~no;L&9mYY{+z<1oopLOnp4+`}y1VS3vtKF_N_4_Gw+Fyp*fF=GA%hdE zF&qnRO^@R2YfKy;jm{mgAe-ZPf}AuPRGWiJln&YQU!weVy)UJSsOeH@()FlA7Z+O_nsTu4%4y#49kM@v(O{-r}Zi zw^Fe`-4FLo9p%cteArwebu~`tP4k*i_DOSjImIN)+`QV-`0_=m$F5F-tkxt|@* zvRu?~}Y>R4q?4#|!8%c*BJ zr}l3aoXq)Y;Iy9!&RpZz`Da<9?4o7bcgl_T%Bf!rr5+s;v}b-OnEQ*tnN(k$52`8r zgmBuHMV?-!JJh%>_3AaLA70aJW7wG_Btan}aLT5i6&lJEpTDH>`KHv7eCmr^1gHPa zr(S1D^zu8-u9Uf8HmjSz+L@Z#d2YHniGH(G_K&Adyh$n*GcG<0mt{J4#2W99rM~yR z)UomA*FAkXUF5R6#%AcwqV8O}ukDpnuf89IXN&g6(@Xf*CoI6VOmR|fJRqlz?vqo8 z_Q@bQTVo57vu}vvXD2xgbNw@cAOimx*pJ}8ywGstXzDvR%c&zb%YSr>gz+{B+ZO)> zZcQur?8M8Vqf9+)%%q;UMNZv)3#jA3?{qk9k4P_{CjG{=r$f;!@rOSor}ls7O%tEq z!Lu_yt9`-t9g}kE|Gjz2Gd4FH70XP1LTvG{{P$(GqLgXI9*=n-ipSDD^eesl2Z?USZMrDbU<*1QJfkhtEQc9 zZhlba=g|3$A6Sq&@CiBf{B4l?EOjANe3ss`sQQ#}?3VM5*&IXU7Fcqsvhn4L#sd|p zr#>mCUbr(-!J8`N&Xn*mRYU}2B`GA_r$&}DRxf8V<@KLQHXf=;-TjvcW1ouF|09|G znJXTB+w_WX!)VFIAN^29+LMYk-t-XMyE#$guUmKJ#>>k&bmEEHrQ)ByeYx-ZgSf=M z_xS#}-?QCE-*$i5_NcF?$M@HWr)9psQrz)AbH_{cxUvTnN`lO)!9H|u#~6RDcpuo_ zi-N`}I{6_CW2R_j^ zSpI6k-%@jC{IZC1e~0*9mQOtSt3^XLTBAdo)~Z7r*M*xm4GK}9(9I|J1w+sk=_jJU zVZ*v!zjyaE|9+YOx$+C5zR*oiE*N1MYJl#@a&@yl9FmQL{+fZ^OPB8M^`_-tR+e8V zXxNcjis;y}6si|r{Yd_v9ZSRLy~i)hN5!W(@~sl@SbvzU z*f2xG-#7r(l$VQy(gZ)z`*^fp{bEvC-iVYiT)%mvmZ2ldcl&FuhWtzWX4db$ zDjFo1o?c&&UKYJG(OI^PvOS3R6&dJT6;J+BvHsHCGy3Q)>my>VJS!R{T%~7(_OJ2# zp@WFu!IS!!k$?O8vq&p$Rl5}oOp9c+cL&z`4Mg|G~)eVr0YqvVs_I4;a(>UK0sf)^GGE*cVU%{S1xM@!l(^*X!|n zdSS|uzm@f~>~#GIRrl;*n^s}3GdqKfw_A{!wqxiaVh!(o{#7UUe`Q1@GLrIsM>8UP zT7f_C+~AF1uZzNG1#ZFk0W5Ud)y2@h$Z4+{0GER2!SxZnIrbw4mB;t5y?$t;mDvPc zMxFgvLmp3^k#Aamr`vDjpMlXMY)%b~RgwQ>1^;$0rV4#b<3|}(XYC~dt)9YW_%_?P zd~|4omkUPMs&sV2^^txe&U-hD{nt<9AIieN+5L#@sej!t)yv4~i}58uZlsURe$RLr z$Fca#q@F)w57AcB5icDW(NI&!&t}&5%leCu|0F4(wiw#Kb|?1sddM*XJro5_VZXCt ze(tpSXXszZIy+yZBiMpfSBt5GqQTc0`;D|R$A84e`f+_>@|5js!#LTrcIOaqNnoN--ON-ek<5|&d&hY3GMdX>M)3DM gN=4Y^{G9|R?}uHjk|d&-e$8ItijSaVW_f#5(H+jGDff(W*{V@VK`z~Ygo;o zXa*x>4WkH#As9u+b()n)Mn_1QRsCxeT%%Eg)pY_RFq)AST#<2=L1~OcWdyO-C@O-n zILQ)*h7*LQD4K$k7)GcHMd~z0TWe$)ml@Qcbdn(nTG3=2)o`65F;dYOn)J(I7#(GB z1!ECJCuBsI2_23VN+T3SS7gP0x=?|Ws*EAJD&qu>Yck0iIL#V5MjASbI8GO-EJGm% zq9fohfit?Iuq;YaGJ~*+uE~r&rHGK;Te>5DY z2+|-80yP*}XDFOtbpyjxl+qA;!$QW;H3r8hO_voyV`Yd84IVKR#$dFDIT{ubI)&2; z3vrS)0tYV)TqPJ<(@B|NNmj9^6cQ?;qpYTo8cGuwqcQ}EV1_~wstQi9j?y9m!F5uh zRhA+ULStA4f}vWya#j8LElg@}L)EKRX+9l(Ylh=Q;N12$6#j)Nk{W}9ZW1Ny)>4Wf!b%t%>L z5L^b=WSnt4XfET3jFOs;(<&v|8JR_8Km$R+0Y)*7oFW!AFd8Rygw{y{@XM-<0sb28}A<1g+`>L+Y%m5u^nlk_6mQ3Gsb31(g9LGTGfaTIW9-&{!OEKNX0P#CQc6yzMG$sj~mG2DO%IUdB5Dx=Xj zt<&~{i8x9s7}&+qP%2c20+FzxP{#=sbP^1zVFYBXjLBAZz(N)P%;<-0vYppZOeQGE zBaFpZ7DF{aJwoG(%4!K40Lu`1O;Z3^P(|#L6TmH>h{O>{E8LM%r0NhmN}(ZmG=h^fhC$46C@m}nX)F^OjX?yJgv5Ldr70OogDfKq>;xj~ z2H+E=AxbFAWP7|aLNFT0#Z{Ripjt9g`-V6|Y7`U$l}v~)#6TZmD1bLjA!HPh8O5Mv z&A<&zr%*I0u26^CMluj;O+z7vR0Tm~1jQh{l*&NXcnTr2bse)RA|$Cs;tXsws{mC{ ziZqObFA7-%rzq7hpd#P|Bs&W=Qo(c_kPW+wIvx~iP?=C1FoRN1b1_IT%#bl%K^VZa zPU@Z_nZ{&PBMbuy1%YUUOfjTN>xzboIzP6VM-xXT@|4z73u{G zMT=xu8OkoL5>UQ0#DIvxeyJD=X{-Pg2?}ALIUuO2sVZu9z=SD@7)?Wc(n#p0uqCvr zD5LJjgBS=LoYu!eIx0GJK?K?ubPEPV7|=f%ih#u*+22b$!7@S|w2^6VORX19}kPgrW?3 zWuQ&9r^KU>wkSig(hA0};eu$rnWv?oP zi3j=$_`p~V7zz_hk_f;s zYo)6JGnlMGOH&;wPEyuJUF$_5hz5XBWCkK(C@|~EkV-1F-a-t8nMo&M41wMbb%{kV z6b3XHU!VwTbV+fz21|*za=R$10r-OiL;)?ZwK9x@wBaZXQrWQOA$=7Dg_(+wX#|=N zv<=8*Yrvv4N5gn1&jtgVMTMn?#G%mtp`asDXe0}Upz_MAg)Gb$Dx?KPY9vf|7*qpL zs=yEcNN220g6lZI48%agg4&3%45L8(fZYcW5k%VwF$Sj5B!)|UY{XlOY)HKk@d`9f zQdS}PZpHylR06W5Uhkwy1WgezJyFo@pvE#uzZ62HBq_ttWEF~)B16f7z#D*LaEO9J z+qV-;kpv8SI^-c-2H!9PWiK5Ghx>7~Nmf2|50WA@*c~WEDnhUr34L3KK~#a#k$p1- z=1r2Mp+Z6RhiMMV02P)P8jq;350HtBe6tY701SmOQp2IBqEPz{66zj`Lc@jlP$j); zHm()}<1$H-m^}%GPXmxoDU5EY2H>G#Bn9K2PKR9$=|sYa1Gz|1D3n960D+XmRxT@K zQGfzOUuGfEAq5l#3It3l0B~B?Q9Y&zgPM-w2$%#p1S2X;u8;`O7BwiIEUwr$!%|oy zD=?618li{%5z-tw98_+K)euE(yDlC^0TK#1%Sy)+NgQJp6`G==qB^DmjuSRO*5L|) zs~Uk_swqvEh!>g$yNQluWCTesg zOky+!dKlys)IA-Vu+B!=qCT=mmD!tHk5S=K!BN3+l&x|6zo`y%O#Hu5Ek{Jav(GCx zdsV4ZT-?{&I3{euSjEg@>XuH>e5*5C&YNQF{k~9GR+^RAbbg0bv-G2}pwd+G62`Kj;qTz`69!J!)?_FUXh7z*;?Auv2D zGA$~={{-la3xad6wkSZsD`!;bUwsO)F2wk$-?pGo`=@qG!m{9~HNlmSk%7NVx3{Pr zB6TVXlDZ*Pr1VqOqz}dgN@tToq)DTKr7`?H?{YL)+IP$*jcyz)#WFUj?X^Iua82JRLv>fj;|)Uy4>>446sSXLxQE{ z)q|z))`G4#1Ec}ohu*NjD$<-A4k=}yLn>l}q_ZCfN#i=&rPA?sscxb}B65HCUi|!) z_i*j%lG^-%x3}()4#qt4ibtzR+aCl--?j;rwymrpwT`SRebMla_uKu|q$Rr^dfRb! z>C#E3wC8HDG;a1CuhQ?fck2sw=}vwX>8B$>Qm6N-N;6;k!`uC8h;$?&NP6v%Ln?3J zl=>a0Dt(CsOBet3&|7M&CY`+Uv-i7G0n(Ucn>6t4U`cWXNkmW;X?>R((r+&WNpIi2 z?rnSFmbb<3hu*#~IHY$*hf49QZg_X@4V8|35hnc>dB^)3_$$;6k%sNQ?{(L&A~g@G zE^X-@Bz529l!Q}Nq&t0sB$@opds64xp|1ub8#Q%r}>QR$?CCJm1{$sKgw)eyVA^Fx1=- z|Asj*XM{N${m6Xu^8oVj*=8~~-rN;hV9qGZH+!s% zHfQzUXy)`8Yc@YM$#k9>Z+0tx#q7|y#1s~RJ&{YzhEbX3*@6{j;)!V{J#DrbIC6{` zpZA8D8}Yiyj26wM7p9umuP!fZHvA*gy!obyWWH$z7IZcz#vC&5mTBhMGlDsG?>px1 zqod4{mIdaDFV~rqr(~P!YWFaQY#nFzn_Xg7XH;{jKFe%!ZLN7@!EUqj_$B5u;bSR^ z39>|6@Tuhf`|xoErs*IZ)0Ri#&XnX;^GXosFI+pd#jVZq$9a_HheDAWp*M2cy z{HCutbwz4fwe@oOw(r-KAK9&&Ncss=eZRPD=)NgsL;6ytbSJ#L->w>E6Eats3Dfe; z^$TZ|zx2uX=JfQJ%NDknQbrDSmA$g?^YRIyoA$8ctn%WJmgU%MX4!`$YL~f6=9Jrd zxyxS9nON4K+6D90$79U6FB_I!$V@HYe{GDp;zEdeC z^4Pjk`G%;h@@K-wQWO(pskGoz$^9?Gr#Luj5;U)}ir+ws1AlqL-Ys+%FPC)SCw4o{ z*Z6oa-}~ADKIYRE{C6QA^6BkA;n%-tw0$MN^T1fR z{5GGP`5Hgyt1Y~H!Y)2)cQ4_#9w^k?4K|h!72;Ed2{v1?$S7@jo6Rc>`ic_HOkB*j z$XmxBn0l5Uv0*-+R6jy!wIfH^bgeenn^hZbzKAR_by z7+t(EUg*=XnJ{y9bK&d!dBO*5xD_4&nf@Mc!Q#*YOxU{Ud%p3c<-&)X3E}mAZ}U?R zu7Z%-Y*lKu5Lz!f%WKU>3r)MM;&+uV0C`nF_8LCxW*wnpe3YTcZ!kzGg)N;kXO8gej5mbi^+<95y-tGgc{d?!Kp){U6DD|SpW(4S zwT096SA>MNHHChLAsloq6h{A;F7%l?QP|OA37=Dg6(&w_f1~>T@?; zE!er_*zLgJlGe(J!p%FI|2akz%!ip4Fdie~8RN?tG)@ZneS-AssZigND$~57i}y?8 zbE`?MwEo`XhsXKe%o`zNBk=NSUy6! z`ARKcOvV;z>w$XGj8j#mgw3ycv7`v;M5E)rE>YpW6?MXWJ#TjRo-LXum8_9`DL?h{ zU24!!dUSD&H09!Osr1ZlNnbQiLYn)e-!Ikj)nB-umYLu`3 z+Hha#YDxMmeTB5~>n)PK=#+O1F;aSY<9uJuq8{G!(Sg2!uSff4YP6ILy6#5K^mTi& zm9PKT$Gxa?gtYZ;fpnwFM4qbifp13E7%BMc1=8Sx>(YB0vb~9q+DMD<&5|Y$p5z-g zH(Kfy_o=Vnm~B$P+gqiP2d+qMzHTISyiqLOb{rEkcIQcYk8bX?nLB(vB5z7N6K{B% zC^ji+R&nc1MG?}e=mc-G^OJlLF41=?mXhwSUgM)m*GfOs@8Nx!TrCx(?3SATi}gkp zIeh!KT=iY<=4zcjv##%%1;r8?6J)8g;8V#x4}AWEk6+J#Pu4JAXwW`Mc&%P5q3h3y z!kSlG3Vh29p&+Y=Fmin@fwD~*IvEn5k{2X5hKG;Avz9a~RWeJW1K-Kq($M3>+?bd^UJ z@AGt?A0eFH>JuL99xN>Te4OwwH&ocO!!5M0+FbZ$7$$U`nJnbZ?8nQcxX|aN7~y(4 zBiyWYmLKem6*`89LY4HELd5<=fo$9bdf#C_G_|R}Xij)=pFetn7CyOmoZq&$z3_Xp zP@yC(MR?(}_QJcF+xa#RN#UD&ZG<(m>Ih{sy9)EqHWmIjBwBrme}AV>=(jpus9p3W zUwVEizx{lS@J#qv23J;}O76c8pR|Q(;)TyUi4k|&h!^&H#i=_wiSoxUi)~tFiWi2t zMD1Ul#03}HibqFK;`?uS~eCK=@adz9T;?D96 z@!-DR;=Mhc#NJVecyvv=IOgRf(LFX@9KO>h_D3Z#U}=)Le0Xbd;m=LPafjQAb3RWL zM`v{s8^yI2PnNb5+qCT>p4#ITkGJY9<{s-HPF%u@P1<%A`)x}WnR>~hXE82*bib|G za=su=Sd=Q}?cu~*_eJqhrcZqNo=+Tm$0u$G%oK0>(#0zgAa{D0Sdi0OJl`cve6T1} zygDsGoX{2*e+WaxZPVT2!|!-;)AXL=+y{)PyqhX^8`?(v{&c$d#TUKA?g{Ngqj9Pj zHjNek;5&=EUTH6W{{k-_{0tYZo;&`_WU+fmTd__c#OFdgaoRpX{CN>Cp1jvd{MVq4 zV)xAUV(LCp?4K)$Bdey0%{Z@kAg_@quB62}vI?4-*d0I5$Soc znCEV3U%5k1{>G*3XzxC;eE?T7yMZU*o1t8ZtCV}aQ8c%rr<*I9Hkotn$l&($*vq|h zxsSU`b2n#9In3Q7%H7eg)ZjX;Ud3G~+`OD<{cZ>iPG%YA-hSl+AC!MQxwT;qqs0skn)!$AVPu zTsX^}`SMq;@8Z2KV#c8}9XcdvYV(ySS3%2u_Y1&*9f{xgnR{;C^yVmOap%Uk++)66!Wm8DJ6J0Bd&rG0e5vu1(J^=OytS#fqDSG8_2mxbPU@2HpSX;Cf4!{?WAN#T*6 z6D{v?A$E=HdE!H^dggBK_`SLwqX z&3$-n8n^Jpj@@lKWqV4-@4{iwg7q z^cF=;-!22| z+Z=XFveRXASm&zQ1Y1Bb{AhjJ5%o1Z$Ze^3kh>o6%K&?i!7gfaH;H zm!?`ur^)#nTTuUMkw>4ne4ng0s0KYt3Zv`nilfIXG*y2wP8}XqogRM9O>HS|Ko#Fx zMMbt4+0j6hjBy z`HC7e^#PSID3WS6m!l$EHl}V4I!7Nr(TdudQb51hKZc$;{}1x~{(O33?`u@}!n#!I zz<6rMvYBMCcOJc7qNu8~j#3{6W7M&EHK<9{5o+B8hU#pKqvg3b$T@Kd^sYk=dhUuC zdWYRbZT-MQQMr$(^8>0AZ|M8zm9NIqedE3*sf?fLt9M?Yn#Y|XFE`Dm-~Idnd4FXy zDmkizt|dPtw{EFT@tuR|JvaK(vv!*_wJMQ*W>vEU#ROU6t&zNvdmdQ+W9jQ zy;^0aQE#6N-crXt_wyTt{Dm4Zkzo^)zKLLCU9%6>+4Oyp@9oc;49!fqyRLgqx5rO( zouay=!HXR0-^4XuWW6PU*RNV1SuWhRXhjrp@^=%U-6|X9ecQ$s~CO9YA zmaSUexbd0*TY%f<7zIlJxn-eIi=2PHZdn%G@T-VPm#%dSwq@fX&YMuAx>w0+QXnJO0NHf{$l2?#uxH>Dhc}3= z5Bx0#$kGl#wrD_(H}og6V8EE|yye7q&6`+6YTjsqEf87`MpEp;G4@}MUb;jxU|JgrYwcE7AXe^nckmk)X#7n43J6qM||e0`8QyyCVR*5z#V zZ#RYZ&wgTJ_Fx+X&tc@q{qo?gZE%_gLPZ9F=O8C4z=~QxOcgyx9|#5NNw|U_lGW6~ z19OMw>rlw@6C%^5M8q|RJ8ib;PwlaBu@Ayswz#O!Ske|89{coWLOx6=HvMMn81Az7}0`*tq}a>(PJcXc8Rjf_J5lZGWP}567oPUrce$ z@`32emt|I?tK=X9pHp-Nyq}hRKHW8=9muYHlVZuP&IT8s`7VTOI0QJ{Hp$$&VLEp0M`s1^i=); zXz8iso?p*1%EQx&&gQ#H2Z5r>pNcF+mE7|y`b)WaTE)L!cTF1xDk^_YuvAoX&#xll zALZ+5J*UUnT~mtfpsDi7)6!JQJ-?>^Ye9Tk;g{3xu1OOug_X}LmcmNz`4#@X#8&9J zX^*~RcU_!ecTJdK>8X66u=G@N&#&h%MY}@DJI?4WPS@B4c31H%OG)K~YbmMZ{=JgN zm71w&1peaxr09dHgZAhx0nxhyT< zQbtC{?(T|tA?8liR@SFbN|Gcg9sXLMc9Jwq^7WJ2NK!wkbwB?(pHGt7TSu)Wsjc-5 z{F84#16T6kHc7UYwH}rzNr`!IyHsE;Ynu+Y`I0QYwZ*p|2Di8FYH93lsrIDp;~gc0hw4E#IhRBABSx9ZfYm32AM=NIkQ4~D?Uuez&sRmOaD>wpA+z zxTH4zl@Kh8u}Hd-5yzL|ESpX=V8J`J_VsS^rHH`uw~_By3oBoy@rC*}_r+(jv zW{>Sp^c(FD;#2nf{qc>^lG`9XyRlh4@Qhpq{$o!Mj2)_CIdHz)&>r{zziG(@^kaK` zYO8|3_~&2nlz$o44QfgU*rTl6o>-RAb5QDI*#unib^i1I)GmLs&#&sK_*zc-@_Tk~ zGc*(HHdISVEl^V`?8*Amp2W6)%BzU?-|gvaNmcNkIS`{fr42g*Vur#oB6nwp0*B{wEg> zP?87x`2xU5h5xV!k_!q0+@f0fYFSbV!pHIXd%h?7{b4bPO_q{eXsB3+0(D?lt6qWY zJ*@AFPkLhdWB(+dA9+^w@vC^eK{PWbm-~#}8)l b7~T?TiNbRPJi7gV{jo)a*=(KQ|0Dh%1)_Ii literal 0 HcmV?d00001 diff --git a/forge/catalog/cli.py b/forge/catalog/cli.py new file mode 100644 index 0000000..1092bdc --- /dev/null +++ b/forge/catalog/cli.py @@ -0,0 +1,244 @@ +"""Catalog CLI commands, wired into the main ``forge`` app. + +Kept import-light at module load (only typer/rich/stdlib) so that wiring these +commands into the base CLI never imports pyarrow/duckdb. Heavy imports and a +clear missing-dependency error live inside the command bodies. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import typer +from rich.console import Console +from rich.table import Table + +if TYPE_CHECKING: + pass + +console = Console() + +catalog_app = typer.Typer( + name="catalog", + help="The catalog — an append-only, queryable registry of ingested episodes.", +) + + +def _require_catalog_deps() -> None: + """Raise a clear install error if the catalog extra isn't installed.""" + missing = [] + for mod in ("pyarrow", "duckdb", "xxhash"): + try: + __import__(mod) + except ImportError: + missing.append(mod) + if missing: + from forge.core.exceptions import MissingDependencyError + + err = MissingDependencyError( + dependency=", ".join(missing), + feature="the Forge catalog", + install_hint="pip install forge-robotics[catalog]", + ) + console.print(f"[red]Error:[/red] {err}") + raise typer.Exit(1) + + +@catalog_app.command("init") +def catalog_init_cmd( + path: str = typer.Argument(..., help="Catalog root (local path or s3://, gs:// URI)"), + exist_ok: bool = typer.Option( + False, "--exist-ok", help="Reuse an existing catalog instead of failing" + ), +) -> None: + """Create a new catalog (writes catalog.json). + + Examples: + forge catalog init ./forge-catalog + forge catalog init s3://lab-bucket/forge-catalog + """ + _require_catalog_deps() + from forge.catalog import Catalog + from forge.core.exceptions import ForgeError + + try: + cat = Catalog.init(path, exist_ok=exist_ok) + except ForgeError as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(1) + console.print( + f"[green]Initialized catalog[/green] at [cyan]{path}[/cyan] " + f"(schema v{cat.schema_version})" + ) + + +@catalog_app.command("stats") +def catalog_stats_cmd( + catalog: str = typer.Option(..., "--catalog", "-c", help="Catalog root"), +) -> None: + """Show catalog summary statistics (counts, tasks, robots, score spread).""" + _require_catalog_deps() + from forge.catalog import Catalog + from forge.core.exceptions import ForgeError + + try: + cat = Catalog.open(catalog) + s = cat.stats() + except ForgeError as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(1) + + console.print(f"[bold]Catalog:[/bold] {catalog}") + console.print( + f" Episodes: [cyan]{s['episodes']}[/cyan] " + f"Frames: [cyan]{s['total_frames']:,}[/cyan] " + f"Hours: [cyan]{s['total_hours']}[/cyan]" + ) + + score = s["overall_score"] + if score.get("scored"): + console.print( + f" Quality (latest): mean [cyan]{score['mean']:.2f}[/cyan] " + f"median [cyan]{score['median']:.2f}[/cyan] " + f"range [dim]{score['min']:.2f}–{score['max']:.2f}[/dim] " + f"over {score['scored']} scored" + ) + + def _print_counts(title: str, rows: list, key: str) -> None: + if not rows: + return + table = Table(title=title, show_header=True, header_style="bold") + table.add_column(key) + table.add_column("episodes", justify="right") + for r in rows[:20]: + table.add_row(str(r[key]), str(r["n"])) + console.print(table) + + _print_counts("By task", s["per_task"], "task") + _print_counts("By robot", s["per_robot"], "robot") + + +def ingest_cmd( + sources: list[str] = typer.Argument( + ..., help="Dataset URIs to ingest (local paths, hf://, s3://, gs://)" + ), + catalog: str = typer.Option(..., "--catalog", "-c", help="Catalog root"), + batch_id: str = typer.Option( + None, "--batch-id", help="Group this run's files (generated if omitted)" + ), + batch_size: int = typer.Option( + 200, "--batch-size", help="Flush staged rows every N new episodes" + ), + strict: bool = typer.Option( + False, "--strict", help="Fail on the first episode error instead of skipping" + ), +) -> None: + """Ingest datasets into a catalog: register + quality-score each episode. + + Re-running over the same sources is a no-op (episodes are skipped by content + hash). + + Examples: + forge ingest ./my_dataset --catalog ./forge-catalog + forge ingest s3://lab-bucket/raw/2026-07-18/ -c s3://lab-bucket/forge-catalog + """ + _require_catalog_deps() + from forge.catalog import Catalog + from forge.catalog.ingest import ingest + from forge.core.exceptions import ForgeError + + try: + cat = Catalog.open(catalog) + except ForgeError as e: + console.print(f"[red]Error:[/red] {e}") + console.print("[dim]Run `forge catalog init` first.[/dim]") + raise typer.Exit(1) + + stats = ingest( + list(sources), + cat, + batch_id=batch_id, + batch_size=batch_size, + strict=strict, + console=console, + ) + console.print( + f"[green]Ingest complete[/green] (batch {stats.batch_id[:8]}): " + f"[cyan]{stats.ingested}[/cyan] ingested, " + f"[yellow]{stats.skipped}[/yellow] skipped, " + f"[red]{stats.failed}[/red] failed, " + f"{stats.frames:,} frames" + ) + + +def query_cmd( + sql: str = typer.Argument(..., help="SQL over: episodes, quality_scores, v_latest_quality"), + catalog: str = typer.Option(..., "--catalog", "-c", help="Catalog root"), + output_format: str = typer.Option( + "table", "--format", "-f", help="Output format: table | json | csv" + ), +) -> None: + """Run a read-only SQL query against the catalog. + + Examples: + forge query "SELECT task, count(*) FROM episodes GROUP BY task" -c ./cat + forge query "SELECT * FROM v_latest_quality LIMIT 5" -c ./cat --format json + """ + _require_catalog_deps() + from forge.catalog import Catalog + from forge.core.exceptions import ForgeError + + try: + result = Catalog.open(catalog).sql(sql) + except ForgeError as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(1) + except Exception as e: # surface DuckDB SQL errors cleanly + console.print(f"[red]Query error:[/red] {e}") + raise typer.Exit(1) + + rows = result.to_pylist() + + if output_format == "json": + console.print_json(json.dumps(rows, default=str)) + return + if output_format == "csv": + import csv + import io + + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow(result.column_names) + for r in rows: + writer.writerow([r.get(c) for c in result.column_names]) + # Plain stdout so it pipes cleanly. + print(buf.getvalue(), end="") + return + + if not rows: + console.print("[dim](no rows)[/dim]") + return + table = Table(show_header=True, header_style="bold") + for col in result.column_names: + table.add_column(col) + for r in rows[:1000]: + table.add_row(*[_fmt_cell(r.get(c)) for c in result.column_names]) + console.print(table) + if len(rows) > 1000: + console.print(f"[dim]… {len(rows) - 1000} more rows (use --format json/csv)[/dim]") + + +def _fmt_cell(value) -> str: + if value is None: + return "" + if isinstance(value, float): + return f"{value:.4g}" + return str(value) + + +def register_catalog_cli(app: typer.Typer) -> None: + """Attach catalog commands to the main forge Typer app.""" + app.add_typer(catalog_app, name="catalog") + app.command("ingest")(ingest_cmd) + app.command("query")(query_cmd) diff --git a/forge/catalog/ingest.py b/forge/catalog/ingest.py new file mode 100644 index 0000000..2895683 --- /dev/null +++ b/forge/catalog/ingest.py @@ -0,0 +1,344 @@ +"""Ingestion pipeline: dataset URIs -> registered, scored episodes. + +Per episode: **discover -> content-hash -> skip-if-exists -> register -> score +-> stage**, flushing staged rows to the catalog per batch. Everything reuses +existing Forge internals — the same format readers that power ``forge inspect`` +for metadata, and ``QualityAnalyzer.analyze_episode`` (the engine behind +``forge quality``) for scoring. No metadata extraction or scoring is +reimplemented here. + +Idempotency & resume: an episode's ``content_hash`` (xxh3 of its proprio stream +plus shape signature) is checked against the catalog before any work; re-running +an ingest over the same sources is a no-op. A crash loses at most the current +uncommitted batch, which the next run re-does — no partial rows are ever +committed (episode + quality commit together, manifest last). +""" + +from __future__ import annotations + +import struct +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import TYPE_CHECKING + +import numpy as np + +from forge.catalog.schema import QUALITY_METRIC_COLUMNS, SCORER_VERSION + +if TYPE_CHECKING: + from rich.console import Console + + from forge.catalog.catalog import Catalog + from forge.core.models import Episode + + +@dataclass +class IngestStats: + """Outcome of an ingest run.""" + + batch_id: str + ingested: int = 0 + skipped: int = 0 + failed: int = 0 + frames: int = 0 + sources: list[str] = field(default_factory=list) + + +def _resolve_source(uri: str) -> Path: + """Resolve a dataset URI (local / hf:// / s3:// / gs://) to a local path.""" + from forge.hub import is_hf_url + from forge.io import is_remote_uri, localize + + if is_hf_url(uri): + from forge.hub.download import download_dataset + + return download_dataset(uri) + if is_remote_uri(uri): + return localize(uri) + return Path(uri) + + +def _episode_content_hash(episode: Episode, frames: list) -> str: + """xxh3 hash of an episode's content: proprio stream + shape signature. + + Deterministic and per-episode, so the same episode always yields the same + hash (exact-duplicate skipping). Streams frame proprio (state/action/ + timestamp) rather than raw file bytes — Forge readers expose logical + episodes, not per-episode byte ranges (multiple episodes often share one + parquet/video file). Images are intentionally excluded (expensive to decode; + proprio + shape signature is a strong content fingerprint for teleop data). + """ + import xxhash + + h = xxhash.xxh3_64() + # Shape signature disambiguates episodes with little/no proprio. + h.update( + struct.pack( + " float | None: + """Coerce a metric to a plain float (or None).""" + if value is None: + return None + try: + v = float(value) + except (TypeError, ValueError): + return None + return None if np.isnan(v) else v + + +def _episode_row( + episode: Episode, + frames: list, + *, + content_hash: str, + source_uri: str, + source_format: str, + dataset_info, + batch_id: str, + now: datetime, +) -> dict: + """Build one ``episodes`` row from an Episode and dataset-level fallbacks.""" + num_frames = len(frames) + fps = episode.fps or (dataset_info.inferred_fps if dataset_info else None) or 0.0 + duration_s = (num_frames / fps) if fps else 0.0 + robot = episode.robot_type or ( + dataset_info.inferred_robot_type if dataset_info else None + ) + + state_dim = episode.state_dim + action_dim = episode.action_dim + if state_dim is None and frames and frames[0].state is not None: + state_dim = int(np.asarray(frames[0].state).shape[-1]) + if action_dim is None and frames and frames[0].action is not None: + action_dim = int(np.asarray(frames[0].action).shape[-1]) + + cameras = [ + { + "name": name, + "width": int(cam.width), + "height": int(cam.height), + "fps": float(fps), + } + for name, cam in (episode.cameras or {}).items() + ] + + meta = episode.metadata or {} + extra = {"source_episode_id": str(episode.episode_id)} + if dataset_info and dataset_info.format_version: + extra["format_version"] = str(dataset_info.format_version) + + return { + "episode_id": uuid.uuid4().hex, + "content_hash": content_hash, + "source_uri": source_uri, + "source_format": source_format, + "robot": robot, + "task": meta.get("task"), + "language_instruction": episode.language_instruction, + "operator_id": meta.get("operator_id"), + "scene_id": meta.get("scene_id"), + "num_frames": int(num_frames), + "duration_s": float(duration_s), + "fps": float(fps), + "cameras": cameras, + "state_dim": int(state_dim) if state_dim is not None else None, + "action_dim": int(action_dim) if action_dim is not None else None, + "ingest_batch_id": batch_id, + "ingested_at": now, + "extra": extra, + } + + +def _quality_row(analyzer, episode: Episode, catalog_episode_id: str, now: datetime) -> dict: + """Score an episode with the shared QualityAnalyzer and map to a row.""" + eq = analyzer.analyze_episode(episode) + row = { + "episode_id": catalog_episode_id, + "scorer_version": SCORER_VERSION, + "overall_score": _f(eq.overall_score), + "flags": list(eq.flags or []), + "computed_at": now, + } + # Mirror EpisodeQuality's flat metric fields onto the schema columns. + metric_values = { + "ldlj": eq.ldlj, + "sparc": eq.sparc, + "dead_fraction": eq.dead_fraction, + "gripper_chatter_rate": eq.gripper_chatter_rate, + "joint_path_length": eq.joint_path_length, + "overall_saturation": eq.overall_saturation, + "mean_entropy": eq.mean_entropy, + "psd_high_fraction": eq.psd_high_fraction, + "state_action_consistency": eq.state_action_consistency, + "static_fraction": eq.static.static_fraction if eq.static else None, + "jitter_ratio": eq.timestamps.jitter_ratio if eq.timestamps else None, + } + for col in QUALITY_METRIC_COLUMNS: + row[col] = _f(metric_values[col]) + return row + + +def ingest( + sources: list[str], + catalog: Catalog, + *, + batch_id: str | None = None, + batch_size: int = 200, + strict: bool = False, + console: Console | None = None, +) -> IngestStats: + """Ingest one or more dataset URIs into ``catalog``. + + Args: + sources: Dataset URIs (local paths, ``hf://``, ``s3://``, ``gs://``). + catalog: An open :class:`~forge.catalog.catalog.Catalog`. + batch_id: Groups this run's committed files; generated if omitted. + batch_size: Flush staged rows every N new episodes. + strict: Re-raise per-episode errors instead of counting and continuing. + console: Optional rich console for a progress bar. + """ + from forge.formats.registry import FormatRegistry + from forge.quality import QualityAnalyzer + + batch_id = batch_id or uuid.uuid4().hex + stats = IngestStats(batch_id=batch_id, sources=list(sources)) + analyzer = QualityAnalyzer() + + # Load existing hashes once; extend as we commit so intra-run dups are + # skipped too. This is the resume mechanism: already-committed episodes + # never get re-scored. + seen_hashes = catalog.episode_hashes() + + staged_eps: list[dict] = [] + staged_q: list[dict] = [] + + progress = None + task = None + if console is not None: + try: + from rich.progress import ( + BarColumn, + MofNCompleteColumn, + Progress, + SpinnerColumn, + TextColumn, + ) + + total = 0 + for src in sources: + try: + path = _resolve_source(src) + fmt = FormatRegistry.detect_format(path) + total += FormatRegistry.get_reader(fmt).inspect(path).num_episodes or 0 + except Exception: + total = 0 + break + progress = Progress( + SpinnerColumn(), + TextColumn("[cyan]Ingesting[/cyan]"), + BarColumn(), + MofNCompleteColumn(), + console=console, + ) + task = progress.add_task("ingest", total=total or None) + progress.start() + except Exception: + progress = None + + def _flush() -> None: + if not staged_eps and not staged_q: + return + catalog.commit_batch( + episodes=staged_eps, quality_scores=staged_q, batch_id=batch_id + ) + catalog._writer.write_checkpoint( + batch_id, + { + "batch_id": batch_id, + "ingested": stats.ingested, + "frames": stats.frames, + "updated_at": datetime.now(timezone.utc).isoformat(), + }, + ) + staged_eps.clear() + staged_q.clear() + + try: + for src in sources: + try: + path = _resolve_source(src) + fmt = FormatRegistry.detect_format(path) + reader = FormatRegistry.get_reader(fmt) + dataset_info = None + try: + dataset_info = reader.inspect(path) + except Exception: + dataset_info = None + + for episode in reader.read_episodes(path): + now = datetime.now(timezone.utc) + try: + frames = episode.load_frames() + content_hash = _episode_content_hash(episode, frames) + if content_hash in seen_hashes: + stats.skipped += 1 + if progress is not None: + progress.advance(task) + continue + + ep_row = _episode_row( + episode, + frames, + content_hash=content_hash, + source_uri=src, + source_format=fmt, + dataset_info=dataset_info, + batch_id=batch_id, + now=now, + ) + q_row = _quality_row(analyzer, episode, ep_row["episode_id"], now) + + staged_eps.append(ep_row) + staged_q.append(q_row) + seen_hashes.add(content_hash) + stats.ingested += 1 + stats.frames += ep_row["num_frames"] + + if len(staged_eps) >= batch_size: + _flush() + except Exception: + if strict: + raise + stats.failed += 1 + finally: + episode.clear_cache() + if progress is not None: + progress.advance(task) + except Exception: + if strict: + raise + stats.failed += 1 + _flush() + finally: + if progress is not None: + progress.stop() + + return stats diff --git a/forge/catalog/schema.py b/forge/catalog/schema.py new file mode 100644 index 0000000..42bbf73 --- /dev/null +++ b/forge/catalog/schema.py @@ -0,0 +1,130 @@ +"""Table schemas for the Forge catalog. + +The catalog is a set of append-only Parquet tables. This module is the single +source of truth for their pyarrow schemas, the catalog schema version, and the +current quality scorer version. Every write goes through :func:`validate_rows`, +which builds a validated ``pyarrow.Table`` and fails loudly on any mismatch. + +Phase 1 defines exactly two tables: ``episodes`` (the registry) and +``quality_scores`` (versioned derived metrics). Later phases (embeddings, dedup, +curation, snapshots) are intentionally absent. +""" + +from __future__ import annotations + +import pyarrow as pa + +from forge.core.exceptions import ForgeError + +# Bump when the catalog table schemas change in a breaking way. catalog.json +# records this; Forge refuses to open a catalog written by a newer version. +SCHEMA_VERSION = 1 + +# Version tag stamped on every quality_scores row. The quality module has no +# version of its own today, so the catalog owns this constant. Bump it when the +# scoring math changes so old scores remain distinguishable and reproducible. +SCORER_VERSION = "q-v1" + +# Names of the per-episode quality metrics, mirrored EXACTLY from the flat dict +# produced by forge.quality.models.EpisodeQuality.to_dict(). Kept as a list so +# schema, validation, and the ingest mapping never drift apart. +QUALITY_METRIC_COLUMNS = [ + "ldlj", + "sparc", + "dead_fraction", + "gripper_chatter_rate", + "joint_path_length", + "overall_saturation", + "mean_entropy", + "psd_high_fraction", + "state_action_consistency", + "static_fraction", + "jitter_ratio", +] + +_CAMERA_STRUCT = pa.struct( + [ + pa.field("name", pa.string(), nullable=False), + pa.field("width", pa.int32()), + pa.field("height", pa.int32()), + pa.field("fps", pa.float32()), + ] +) + +# One row per ingested episode. Facts only — nothing derived, nothing mutable. +EPISODES_SCHEMA = pa.schema( + [ + pa.field("episode_id", pa.string(), nullable=False), + pa.field("content_hash", pa.string(), nullable=False), + pa.field("source_uri", pa.string(), nullable=False), + pa.field("source_format", pa.string(), nullable=False), + pa.field("robot", pa.string()), + pa.field("task", pa.string()), + pa.field("language_instruction", pa.string()), + pa.field("operator_id", pa.string()), + pa.field("scene_id", pa.string()), + pa.field("num_frames", pa.int32(), nullable=False), + pa.field("duration_s", pa.float32(), nullable=False), + pa.field("fps", pa.float32(), nullable=False), + pa.field("cameras", pa.list_(_CAMERA_STRUCT), nullable=False), + pa.field("state_dim", pa.int16()), + pa.field("action_dim", pa.int16()), + pa.field("ingest_batch_id", pa.string(), nullable=False), + pa.field("ingested_at", pa.timestamp("us", tz="UTC"), nullable=False), + pa.field("extra", pa.map_(pa.string(), pa.string()), nullable=False), + ] +) + +# One row per (episode_id, scorer_version). +QUALITY_SCORES_SCHEMA = pa.schema( + [ + pa.field("episode_id", pa.string(), nullable=False), + pa.field("scorer_version", pa.string(), nullable=False), + pa.field("overall_score", pa.float32()), + *[pa.field(name, pa.float32()) for name in QUALITY_METRIC_COLUMNS], + pa.field("flags", pa.list_(pa.string()), nullable=False), + pa.field("computed_at", pa.timestamp("us", tz="UTC"), nullable=False), + ] +) + +# Logical table name -> (pyarrow schema, partition column). +TABLES: dict[str, tuple[pa.Schema, str]] = { + "episodes": (EPISODES_SCHEMA, "ingest_date"), + "quality_scores": (QUALITY_SCORES_SCHEMA, "ingest_date"), +} + + +class CatalogSchemaError(ForgeError): + """Raised when rows don't conform to a catalog table schema.""" + + +def validate_rows(rows: list[dict], schema: pa.Schema, *, table: str) -> pa.Table: + """Build a validated ``pyarrow.Table`` from ``rows`` for ``table``. + + Fails loudly (``CatalogSchemaError``) on any mismatch: unexpected columns, + a required (non-nullable) field that is missing or null, or a value that + can't be coerced to the declared type. This is the only sanctioned way to + turn Python dicts into catalog Parquet data. + """ + allowed = set(schema.names) + required = {f.name for f in schema if not f.nullable} + + for i, row in enumerate(rows): + extra_keys = set(row) - allowed + if extra_keys: + raise CatalogSchemaError( + f"{table}: row {i} has unexpected column(s) " + f"{sorted(extra_keys)}; allowed: {sorted(allowed)}" + ) + missing = {k for k in required if row.get(k) is None} + if missing: + raise CatalogSchemaError( + f"{table}: row {i} missing required value(s) {sorted(missing)}" + ) + + try: + tbl = pa.Table.from_pylist(rows, schema=schema) + tbl.validate(full=True) + except (pa.ArrowInvalid, pa.ArrowTypeError, pa.ArrowNotImplementedError) as exc: + raise CatalogSchemaError(f"{table}: {exc}") from exc + return tbl diff --git a/forge/catalog/writer.py b/forge/catalog/writer.py new file mode 100644 index 0000000..413ad9b --- /dev/null +++ b/forge/catalog/writer.py @@ -0,0 +1,144 @@ +"""The only module that writes files into a catalog directory. + +Enforces the catalog's write invariants: + +* **Append-only.** Every commit creates new ``part-.parquet`` files; no + existing file is ever rewritten. +* **Manifest-last commit protocol.** All part-files for a batch are written + first; the batch ``_manifest.json`` is written LAST and is the commit marker. + A crash before the manifest leaves orphan part-files and no manifest — a + re-run recovers because episodes are skipped by content hash. +* **Atomic file appearance.** On local filesystems each file is written to a + temp name and renamed into place, so a reader never sees a half-written file. + On object stores the object PUT is itself atomic. + +Works over any fsspec filesystem (local, ``s3://``, ``gs://``, ``memory://``) +via :func:`forge.io.paths.get_filesystem`, so catalogs live locally or in the +cloud with identical semantics. +""" + +from __future__ import annotations + +import json +import uuid + +import pyarrow as pa +import pyarrow.parquet as pq + +from forge.catalog.schema import SCHEMA_VERSION +from forge.io.paths import get_filesystem + + +def _normalize_protocol(protocol) -> set[str]: + if isinstance(protocol, str): + return {protocol} + return set(protocol) + + +class CatalogWriter: + """Writes part-files, manifests, and ``catalog.json`` for one catalog root.""" + + def __init__(self, root: str): + self.root_uri = str(root) + self._fs, self._root = get_filesystem(self.root_uri) + self._is_local = bool( + _normalize_protocol(self._fs.protocol) & {"file", "local"} + ) + + # -- path helpers ------------------------------------------------------- + + def _join(self, *parts: str) -> str: + return "/".join([self._root.rstrip("/"), *[p.strip("/") for p in parts]]) + + # -- atomic primitives -------------------------------------------------- + + def _write_bytes_atomic(self, path: str, write_fn) -> None: + """Write via ``write_fn(file_obj)``, atomically on local filesystems. + + ``write_fn`` receives an open binary file handle. On local filesystems + the write goes to a temp sibling and is renamed into place; elsewhere + it's written directly (object PUTs appear atomically). + """ + parent = path.rsplit("/", 1)[0] + try: + self._fs.makedirs(parent, exist_ok=True) + except (FileExistsError, NotImplementedError): + pass + + if self._is_local: + tmp = f"{path}.tmp-{uuid.uuid4().hex}" + try: + with self._fs.open(tmp, "wb") as f: + write_fn(f) + self._fs.mv(tmp, path) + except BaseException: + if self._fs.exists(tmp): + self._fs.rm(tmp) + raise + else: + with self._fs.open(path, "wb") as f: + write_fn(f) + + def _write_json_atomic(self, path: str, payload: dict) -> None: + data = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8") + self._write_bytes_atomic(path, lambda f: f.write(data)) + + # -- public API --------------------------------------------------------- + + def write_catalog_json(self, meta: dict) -> None: + """Write the root ``catalog.json`` (schema version, config, timestamps).""" + self._write_json_atomic(self._join("catalog.json"), meta) + + def write_checkpoint(self, batch_id: str, payload: dict) -> None: + """Write/overwrite a batch checkpoint (progress marker, not table data). + + The only intentionally-mutable file in a catalog: updated after each + flush so a crashed ingest's progress is visible. Correctness of resume + still comes from content-hash skipping, not this file. + """ + self._write_json_atomic( + self._join("_batches", batch_id, "_checkpoint.json"), payload + ) + + def commit_batch( + self, + batch_id: str, + tables: dict[str, pa.Table], + *, + ingest_date: str, + ) -> dict: + """Commit one batch: write all part-files, then the manifest LAST. + + Args: + batch_id: Groups this ingest run's files. + tables: Validated ``pyarrow.Table`` per logical table name. Empty + tables are skipped. + ingest_date: UTC ``YYYY-MM-DD`` partition value for this batch. + + Returns: + The manifest dict that was committed. + """ + produced: list[dict] = [] + for table_name, table in tables.items(): + if table.num_rows == 0: + continue + rel = f"{table_name}/ingest_date={ingest_date}/part-{uuid.uuid4().hex}.parquet" + self._write_bytes_atomic( + self._join(rel), lambda f, t=table: pq.write_table(t, f) + ) + produced.append({"table": table_name, "path": rel, "rows": table.num_rows}) + + manifest = { + "batch_id": batch_id, + "schema_version": SCHEMA_VERSION, + "ingest_date": ingest_date, + "files": produced, + } + # The manifest is the commit marker for THIS flush — written strictly + # last, under a unique name so multiple flushes in one batch each get + # their own atomic commit marker (readers trust only manifested files). + manifest_name = f"manifest-{uuid.uuid4().hex}.json" + self._write_json_atomic( + self._join("_batches", batch_id, manifest_name), manifest + ) + return manifest diff --git a/forge/cli.py b/forge/cli.py index 2b61f21..5bd72e6 100644 --- a/forge/cli.py +++ b/forge/cli.py @@ -27,6 +27,14 @@ help="Action tokenizers - discretize action streams into tokens.", ) app.add_typer(tokenize_app, name="tokenize") + +# Catalog commands (forge catalog init/stats, forge ingest, forge query). Wired +# via a lightweight helper so the base CLI never imports pyarrow/duckdb — those +# load only when a catalog command actually runs. +from forge.catalog.cli import register_catalog_cli # noqa: E402 + +register_catalog_cli(app) + console = Console() diff --git a/pyproject.toml b/pyproject.toml index 279b617..8e7aaa9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,11 @@ lerobot = [ hub = [ "huggingface_hub>=0.20.0", ] +catalog = [ + "pyarrow>=14.0.0", + "duckdb>=0.10.0", + "xxhash>=3.0.0", +] s3 = [ "s3fs>=2023.9.0", ] @@ -92,7 +97,7 @@ tokenize-learned = [ "scikit-learn>=1.3.0", ] all = [ - "forge-robotics[rlds,lerobot,zarr,rosbag,mcap,hdf5,robodm,video,hub,s3,gcs,visualize,registry,segment,rerun,tokenize-fast,tokenize-learned]", + "forge-robotics[rlds,lerobot,zarr,rosbag,mcap,hdf5,robodm,video,hub,s3,gcs,catalog,visualize,registry,segment,rerun,tokenize-fast,tokenize-learned]", ] dev = [ "pytest>=7.4.0", @@ -101,6 +106,9 @@ dev = [ "ruff>=0.1.0", "moto[s3,server]>=4.2.0", "s3fs>=2023.9.0", + "duckdb>=0.10.0", + "xxhash>=3.0.0", + "pyarrow>=14.0.0", ] [project.scripts] @@ -116,6 +124,9 @@ Issues = "https://github.com/arpitg1304/forge/issues" packages = ["forge"] # Include bundled data files (MCAP schema definitions) in the wheel. include = ["forge/formats/mcap/schemas/*.msg"] +# The example catalog lives in the repo for docs/demos but must NOT ship in the +# distributed package (it's demo Parquet data, not library code). +exclude = ["forge/catalog/catalog_example_droid_100"] [tool.hatch.build.targets.sdist] include = [ @@ -128,6 +139,7 @@ include = [ "/LICENSE", "/pyproject.toml", ] +exclude = ["forge/catalog/catalog_example_droid_100"] [tool.ruff] line-length = 100 diff --git a/tests/test_catalog.py b/tests/test_catalog.py new file mode 100644 index 0000000..0fd5883 --- /dev/null +++ b/tests/test_catalog.py @@ -0,0 +1,398 @@ +"""Tests for the Forge catalog (Phase 1). + +Covers schema validation, the manifest-last commit protocol and crash recovery, +hash-based idempotent re-ingest, ``v_latest_quality`` version selection, an +end-to-end ingest→query→stats integration on a synthetic dataset, and catalog +read/write over fsspec's ``memory://`` filesystem (the cloud stand-in). +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +# The catalog is an optional extra; skip the whole module if it's not installed. +pytest.importorskip("pyarrow") +pytest.importorskip("duckdb") +pytest.importorskip("xxhash") + +from forge.catalog import Catalog, CatalogError # noqa: E402 +from forge.catalog.schema import ( # noqa: E402 + EPISODES_SCHEMA, + QUALITY_SCORES_SCHEMA, + SCHEMA_VERSION, + SCORER_VERSION, + CatalogSchemaError, + validate_rows, +) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _ep_row(**over) -> dict: + row = { + "episode_id": "ep-1", + "content_hash": "hash-1", + "source_uri": "/data/ds", + "source_format": "lerobot-v3", + "robot": "franka", + "task": "pick", + "language_instruction": "pick up the cube", + "operator_id": None, + "scene_id": None, + "num_frames": 100, + "duration_s": 6.6, + "fps": 15.0, + "cameras": [{"name": "wrist", "width": 320, "height": 180, "fps": 15.0}], + "state_dim": 7, + "action_dim": 7, + "ingest_batch_id": "batch-1", + "ingested_at": _now(), + "extra": {"source_episode_id": "0"}, + } + row.update(over) + return row + + +def _q_row(episode_id="ep-1", scorer_version=SCORER_VERSION, overall_score=8.0, **over) -> dict: + row = { + "episode_id": episode_id, + "scorer_version": scorer_version, + "overall_score": overall_score, + "ldlj": 0.6, + "sparc": -13.4, + "dead_fraction": 0.02, + "gripper_chatter_rate": 0.0, + "joint_path_length": 1.2, + "overall_saturation": 0.82, + "mean_entropy": 0.69, + "psd_high_fraction": 0.003, + "state_action_consistency": 0.003, + "static_fraction": 0.0, + "jitter_ratio": 0.01, + "flags": [], + "computed_at": _now(), + } + row.update(over) + return row + + +# --------------------------------------------------------------------------- +# Schema validation +# --------------------------------------------------------------------------- + + +class TestSchemaValidation: + def test_valid_rows_build_table(self): + tbl = validate_rows([_ep_row()], EPISODES_SCHEMA, table="episodes") + assert tbl.num_rows == 1 + assert tbl.schema.equals(EPISODES_SCHEMA) + + def test_rejects_unexpected_column(self): + with pytest.raises(CatalogSchemaError, match="unexpected column"): + validate_rows([_ep_row(bogus="x")], EPISODES_SCHEMA, table="episodes") + + def test_rejects_missing_required(self): + row = _ep_row() + del row["source_uri"] + with pytest.raises(CatalogSchemaError, match="missing required"): + validate_rows([row], EPISODES_SCHEMA, table="episodes") + + def test_rejects_null_in_required(self): + with pytest.raises(CatalogSchemaError, match="missing required"): + validate_rows([_ep_row(num_frames=None)], EPISODES_SCHEMA, table="episodes") + + def test_rejects_bad_type(self): + with pytest.raises(CatalogSchemaError): + validate_rows( + [_ep_row(num_frames="not-an-int")], EPISODES_SCHEMA, table="episodes" + ) + + def test_quality_schema_mirrors_metric_names(self): + # Guardrail: the schema columns must match EpisodeQuality's flat metrics. + names = set(QUALITY_SCORES_SCHEMA.names) + for metric in ("ldlj", "sparc", "dead_fraction", "static_fraction", "jitter_ratio"): + assert metric in names + + +# --------------------------------------------------------------------------- +# Lifecycle, commit & read +# --------------------------------------------------------------------------- + + +class TestLifecycle: + def test_init_creates_catalog_json(self, tmp_path: Path): + cat = Catalog.init(str(tmp_path / "cat")) + assert cat.schema_version == SCHEMA_VERSION + assert (tmp_path / "cat" / "catalog.json").exists() + + def test_open_missing_raises(self, tmp_path: Path): + with pytest.raises(CatalogError, match="No catalog"): + Catalog.open(str(tmp_path / "nope")) + + def test_init_refuses_existing(self, tmp_path: Path): + root = str(tmp_path / "cat") + Catalog.init(root) + with pytest.raises(CatalogError, match="already exists"): + Catalog.init(root) + # exist_ok reuses it + assert Catalog.init(root, exist_ok=True).schema_version == SCHEMA_VERSION + + def test_refuses_newer_schema_version(self, tmp_path: Path): + import json + + root = tmp_path / "cat" + root.mkdir() + (root / "catalog.json").write_text(json.dumps({"schema_version": 999})) + with pytest.raises(CatalogError, match="schema_version 999"): + Catalog.open(str(root)) + + def test_empty_catalog_queries_and_stats(self, tmp_path: Path): + cat = Catalog.init(str(tmp_path / "cat")) + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 0 + assert cat.stats()["episodes"] == 0 + # views exist and are joinable even when empty + assert cat.sql("SELECT count(*) n FROM v_latest_quality").to_pylist()[0]["n"] == 0 + + +class TestCommitAndRead: + def test_commit_and_query(self, tmp_path: Path): + cat = Catalog.init(str(tmp_path / "cat")) + cat.commit_batch(episodes=[_ep_row()], quality_scores=[_q_row()], batch_id="b1") + + rows = cat.sql( + "SELECT e.robot, e.task, q.overall_score " + "FROM episodes e JOIN v_latest_quality q USING(episode_id)" + ).to_pylist() + assert rows == [{"robot": "franka", "task": "pick", "overall_score": pytest.approx(8.0)}] + + def test_v_latest_quality_picks_newest_version(self, tmp_path: Path): + cat = Catalog.init(str(tmp_path / "cat")) + cat.commit_batch( + episodes=[_ep_row()], + quality_scores=[_q_row(scorer_version="q-v1", overall_score=6.0)], + batch_id="b1", + ) + # Re-score with a newer version: both rows persist, latest wins. + cat.add_quality_scores( + [_q_row(scorer_version="q-v2", overall_score=9.0)], batch_id="b2" + ) + assert cat.sql("SELECT count(*) n FROM quality_scores").to_pylist()[0]["n"] == 2 + latest = cat.sql( + "SELECT scorer_version, overall_score FROM v_latest_quality" + ).to_pylist() + assert latest == [{"scorer_version": "q-v2", "overall_score": pytest.approx(9.0)}] + + def test_stats(self, tmp_path: Path): + cat = Catalog.init(str(tmp_path / "cat")) + cat.commit_batch( + episodes=[ + _ep_row(episode_id="a", content_hash="h-a", task="pick", robot="franka", num_frames=100), + _ep_row(episode_id="b", content_hash="h-b", task="place", robot="franka", num_frames=200), + ], + quality_scores=[ + _q_row(episode_id="a", overall_score=6.0), + _q_row(episode_id="b", overall_score=8.0), + ], + batch_id="b1", + ) + s = cat.stats() + assert s["episodes"] == 2 + assert s["total_frames"] == 300 + assert {r["task"] for r in s["per_task"]} == {"pick", "place"} + assert s["overall_score"]["mean"] == pytest.approx(7.0) + + +# --------------------------------------------------------------------------- +# Commit protocol & crash recovery +# --------------------------------------------------------------------------- + + +class TestCommitProtocol: + def test_crash_before_manifest_leaves_no_visible_rows(self, tmp_path: Path, monkeypatch): + cat = Catalog.init(str(tmp_path / "cat")) + writer = cat._writer + real = writer._write_json_atomic + + def boom(path, payload): + if "manifest-" in path: + raise RuntimeError("simulated crash before manifest") + return real(path, payload) + + monkeypatch.setattr(writer, "_write_json_atomic", boom) + with pytest.raises(RuntimeError): + cat.commit_batch( + episodes=[_ep_row()], quality_scores=[_q_row()], batch_id="crashed" + ) + + # Part-files exist on disk but no manifest committed them. + assert cat.orphan_part_files(), "expected orphan part-files from the crash" + # ...so they are invisible to queries (atomic: no partial rows). + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 0 + + def test_recovers_after_crash(self, tmp_path: Path, monkeypatch): + cat = Catalog.init(str(tmp_path / "cat")) + writer = cat._writer + real = writer._write_json_atomic + + def boom(path, payload): + if "manifest-" in path: + raise RuntimeError("crash") + return real(path, payload) + + monkeypatch.setattr(writer, "_write_json_atomic", boom) + with pytest.raises(RuntimeError): + cat.commit_batch(episodes=[_ep_row()], quality_scores=[_q_row()], batch_id="crashed") + + # Restore and re-commit properly — the retry succeeds and is visible. + monkeypatch.setattr(writer, "_write_json_atomic", real) + cat.commit_batch(episodes=[_ep_row()], quality_scores=[_q_row()], batch_id="retry") + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 1 + + +# --------------------------------------------------------------------------- +# Ingest integration (synthetic zarr dataset) +# --------------------------------------------------------------------------- + + +def _make_zarr_dataset(path: Path, *, episodes: int = 2, frames: int = 20) -> Path: + import numpy as np + + zarr = pytest.importorskip("zarr") + root = zarr.open(str(path), mode="w") + root.attrs["fps"] = 10 + total = episodes * frames + + data = root.create_group("data") + dmk = getattr(data, "create_array", None) or data.create_dataset + action = dmk("action", shape=(total, 7), dtype="float32") + action[:] = np.random.RandomState(0).randn(total, 7).astype("float32") + state = dmk("robot_state", shape=(total, 7), dtype="float32") + state[:] = np.random.RandomState(1).randn(total, 7).astype("float32") + + meta = root.create_group("meta") + mmk = getattr(meta, "create_array", None) or meta.create_dataset + ends = mmk("episode_ends", shape=(episodes,), dtype="int64") + ends[:] = np.array([(i + 1) * frames for i in range(episodes)], dtype="int64") + return path + + +class TestIngest: + def test_ingest_registers_and_scores(self, tmp_path: Path): + from forge.catalog.ingest import ingest + + ds = _make_zarr_dataset(tmp_path / "ds.zarr", episodes=3, frames=15) + cat = Catalog.init(str(tmp_path / "cat")) + stats = ingest([str(ds)], cat) + + assert stats.ingested == 3 + assert stats.skipped == 0 + assert stats.frames == 45 + # Every episode got a quality score row. + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 3 + assert cat.sql("SELECT count(*) n FROM v_latest_quality").to_pylist()[0]["n"] == 3 + scored = cat.sql( + "SELECT count(*) n FROM v_latest_quality WHERE overall_score IS NOT NULL" + ).to_pylist()[0]["n"] + assert scored == 3 + # source_format was detected and recorded. + fmts = cat.sql("SELECT DISTINCT source_format FROM episodes").to_pylist() + assert fmts == [{"source_format": "zarr"}] + + def test_idempotent_reingest(self, tmp_path: Path): + from forge.catalog.ingest import ingest + + ds = _make_zarr_dataset(tmp_path / "ds.zarr", episodes=2, frames=10) + cat = Catalog.init(str(tmp_path / "cat")) + + first = ingest([str(ds)], cat) + assert first.ingested == 2 + + second = ingest([str(ds)], cat) + assert second.ingested == 0 + assert second.skipped == 2 + # No duplicate rows created. + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 2 + + +# --------------------------------------------------------------------------- +# Cloud (fsspec memory://) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def clean_memory_fs(): + import fsspec + + fs = fsspec.filesystem("memory") + try: + for p in fs.find("/"): + fs.rm(p) + except FileNotFoundError: + pass + yield fs + + +class TestCloudMemory: + def test_catalog_over_memory(self, clean_memory_fs): + cat = Catalog.init("memory://labcat") + cat.commit_batch(episodes=[_ep_row()], quality_scores=[_q_row()], batch_id="b1") + + assert cat.sql("SELECT count(*) n FROM episodes").to_pylist()[0]["n"] == 1 + # Reopen from a fresh handle — state is durable on the (memory) fs. + reopened = Catalog.open("memory://labcat") + assert reopened.stats()["episodes"] == 1 + + def test_content_hash_dedup_over_memory(self, clean_memory_fs): + cat = Catalog.init("memory://labcat2") + cat.commit_batch(episodes=[_ep_row(content_hash="dup")], batch_id="b1") + assert cat.episode_hashes() == {"dup"} + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +class TestCatalogCLI: + def test_init_ingest_query_stats(self, tmp_path: Path): + from typer.testing import CliRunner + + from forge.cli import app + + runner = CliRunner() + ds = _make_zarr_dataset(tmp_path / "ds.zarr", episodes=2, frames=12) + cat_root = str(tmp_path / "cat") + + r = runner.invoke(app, ["catalog", "init", cat_root]) + assert r.exit_code == 0, r.output + + r = runner.invoke(app, ["ingest", str(ds), "--catalog", cat_root]) + assert r.exit_code == 0, r.output + assert "2 ingested" in r.output + + r = runner.invoke( + app, + ["query", "SELECT count(*) n FROM episodes", "-c", cat_root, "-f", "json"], + ) + assert r.exit_code == 0, r.output + assert '"n": 2' in r.output + + r = runner.invoke(app, ["catalog", "stats", "--catalog", cat_root]) + assert r.exit_code == 0, r.output + assert "Episodes: 2" in r.output + + def test_query_missing_catalog_errors_cleanly(self, tmp_path: Path): + from typer.testing import CliRunner + + from forge.cli import app + + r = CliRunner().invoke( + app, ["query", "SELECT 1", "-c", str(tmp_path / "nope")] + ) + assert r.exit_code == 1 + assert "No catalog" in r.output