feat(datagen): stream writer, resume, item tree, load-blob across all layers - #219
Merged
Merged
Conversation
Bring the datagen checkpoint delta-log in line with the authoritative
spec. This changes the event vocabulary and folded model only; the
storage mechanism (single append-only log.lance, MemWAL sharding,
deterministic event_id, blob offload) is unchanged.
- Event types 6 -> 7: add STEP_STARTED.
- Replace the `terminal` column with a `status` column
(running / completed / filtered / failed).
- Replace step_instance_id + iteration provenance with structured
step_kind / enclosing_step / selector_step.
- Structured DatagenItemId with materialized path `root/step:idx/...`.
- New DatagenStepKind {Root, Leaf, Sequence, Loop, MapReduce, Branch,
SubPipeline, Conditional, Router}; only drivers emit STEP_STARTED.
- fold_datagen_events returns Option (None when no ITEM_CREATED);
FIELD_SET last-writer-wins, FIELD_APPEND accumulates; two read
lenses (lifecycle vs failure). DATAGEN_SCHEMA_VERSION = 2.
- Rewrite specs/datagen-checkpoint-schema.md for schema v2: 7 events, status column, structured item id + step provenance, read lenses. - Add docs/design/using-datagen-store.md: a client-facing walkthrough of open/write/checkpoint/resume/read with runnable snippets. - Add 7 fold + store tests: step_kind/status parse round-trips, filtered terminal, selector_step on the chosen child, fan-out sub-item lineage, set/append mixing rejection, resume open-frame (started minus completed), and a fan-out tree read + root classification through the store. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- lance-context umbrella re-exported the pre-reshape name DatagenTrajectoryPoint; rename to DatagenTrajectory so the crate (and the python wheel + tests that depend on it) compiles again. - allow(large_enum_variant) on DatagenItemLookup (Found is the hot path) and drop a redundant #[must_use] on iter(). - cargo fmt. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… layers The schema-v2 reshape (lance-format#205) landed the data model but not the API layer on top of it. This wires the datagen store through the full stack so clients can append, fold, and read blobs both embedded and over the server: - api: `DatagenStoreApi` trait (RPITIT) + wire DTOs mirroring the Python dicts - core: `impl DatagenStoreApi for DatagenStore` with core<->DTO converters - client: `RemoteDatagenStore` HTTP client - unified: `enum DatagenStore {Local, Remote}` with dispatch - server: `/api/v1/datagen` routes (create/list/get/delete, events, fold, failures, root-item-statuses, blob fetch) - python: PyO3 `DatagenStore` binding + `open`/`connect`/`connect_or_create` Verified end-to-end locally: identical checkpoint round-trips through both the embedded (local-file) and remote (server) paths fold to the same item state. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… layers Complete the DatagenStore surface Xucheng needs for the datagen POC, wired through all six layers (core -> api -> client/server/unified -> PyO3 -> python wrapper) and exposed on the high-level `lance_context.DatagenStore` wrapper. - A1 DatagenStreamWriter: client-side state machine that stamps the bookkeeping columns (item_seq, attempt, checkpoint_id, event_id) and returns event dicts to hand to append/append_checkpoint. Works embedded and remote with no new HTTP endpoint. Fresh via open_stream (attempt=0); resume via resume_stream (attempt=last_attempt+1, continuing item_seq). - A3 resume/attempt>0 fold semantics: resuming_writer rebuilds a writer from the folded item; folds fold across attempts. - A4 load_blob by field name: core load_blob(folded, field_name) resolves the folded item's blob_event_ids map to bytes; propagated to python via the folded dict's blob_event_ids + get_blob, composed in the api.py wrapper. - A5 inspection tree: item_tree folds every projected descendant and links parent->child. api.py high-level wrapper gains open_stream/resume_stream/item_tree/load_blob plus a DatagenStreamWriter wrapper class so callers using the public `lance_context.DatagenStore` reach all of the above. Tests: python/tests/test_datagen.py (6) plus a core store integration test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
YangjunZ
force-pushed
the
datagen/load-blob-by-field
branch
from
July 28, 2026 06:59
3e3a146 to
2b665a9
Compare
# Conflicts: # crates/lance-context-core/src/lib.rs # python/python/lance_context/__init__.py # python/python/lance_context/api.py # python/src/lib.rs
beinan
approved these changes
Jul 28, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Completes the
DatagenStoresurface the datagen POC needs, wired through all six layers (core → api → client/server/unified → PyO3 → python wrapper) and exposed on the high-levellance_context.DatagenStorewrapper so callers using the public Python API reach every new function.Pieces
DatagenStreamWriter: a client-side state machine that stamps the bookkeeping columns (item_seq,attempt,checkpoint_id,event_id) and returns event dicts to hand toappend/append_checkpoint. No new HTTP endpoint — works embedded and remote. Fresh streams viaopen_stream(attempt=0, next_seq=1); resume viaresume_stream(attempt=last_attempt+1, continuingitem_seq).resuming_writerrebuilds a writer from the folded item; the fold folds correctly across attempts.load_blobby field name: coreload_blob(folded, field_name)resolves the folded item'sblob_event_idsmap to bytes. Propagated to Python via the folded dict'sblob_event_ids+get_blob, composed in theapi.pywrapper.item_treefolds every projected descendant to its latest state and links parent→child. Server does a thin rawevents_for_rootdump; the client folds the tree via a single-sourceDatagenItemTree::build.Python API
The high-level
lance_context.DatagenStorewrapper gainsopen_stream/resume_stream/item_tree/load_blob, plus aDatagenStreamWriterwrapper class (step_started,step_completed,item_terminal,item_failed,item_id/attemptproperties). Both are re-exported fromlance_context.Tests
python/tests/test_datagen.py— 6 tests covering open/resume/item_tree/load_blob/item_failed through the public wrapper.item_treelinks roots/children/status/query_tags/fields.All green: Rust 25 datagen core tests, Python 6 tests,
ruff,pyright,cargo fmt --check.Note
Branches from before origin/main merged the StorageBase migration (#216/#217/#218). CI may surface conflicts against those; will rebase/fix as needed.