flaminis/dalaran/dalaran-chunk-processing

Dalaran chunk processing

The pipeline layer between raw data and an DLR: readers produce Chunks, streams transform them, terminal calls execute. This skill is the generic mechanics only. Decide the data model first (dalaran-data-model), then pick the importer skill for each source:

The API is dalaran.experimental; when behavior matters, check the installed surface: python -c "from dalaran.experimental import LazyChunkStream; help(LazyChunkStream)".

Decision rule: where does each component come from?

Default: a reader produces the chunks; lenses shape them. Walk this before writing any conversion code — most "build it by hand" instincts are wrong here:

1. Source a reader supports? Use the reader's .stream(); never hand-parse

and re-log. MCAP→McapReader, URDF→UrdfTree, parquet→ParquetReader, DLR→RrdReader, LeRobot dir→log_file_from_path.

1. A decoder already emits the archetype? Foxglove gives Transform3D,

Pinhole, VideoStream (real sample bytes) ready-made — pass it through, do not re-derive. Only custom-protobuf topics arrive as <Name>:message and need a lens (see dalaran-mcap).

1. Fix an existing component in place (swapped resolution, recolor, unit

convert)? MutateLens, output_mode="forward_unmatched".

1. Derive a new component/entity (FK→/tf, scalars from a message)?

DeriveLens. To scatter one row into N (a joint batch → per-joint /tf), use the two-lens pair: derive the batch with output_mode="forward_all" (keeps the originals, e.g. the joint states), then a second DeriveLens with scatter=True and output_mode="drop_unmatched" (emits only the scattered rows). See the robot_data_preprocessing example.

1. Genuine sidecar no reader or lens can produce (JSON calibration offsets,

hand-measured extrinsics, external metadata)? Chunk.from_columns + from_iter.

1. Finish with LazyChunkStream.merge(...) →

.collect(optimize=OptimizationProfile.OBJECT_STORE) → write_dlr(application_id, recording_id).

Why this order: the pipeline stays lazy, columnar, multithreaded, and OBJECT_STORE-optimizable. A hand-built row loop or an out-of-lens pa.array throws all of that away — that is the path we are deliberately avoiding.

Anti-patterns (use a reader + lens instead)

If you are writing the left, stop and use the right:

• `for`-loop building rows/components → a lens with a Selector(...).pipe(...)

PyArrow-compute callback.

• `rr.init` + `rr.log` per message for conversion → that is _live_ logging;

for ingestion, read with a reader and write_dlr.

• **chunk.to_record_batch() + pc.filter then rebuilding via

Chunk.from_columns** (row-thinning by hand) → stream.drop(content=...), .split(...), or a MutateLens returning a filtered pa.array.

• `pa.array` / `pa.RecordBatch` / `np.frombuffer` assembled OUTSIDE a lens →

move the transform inside a MutateLens/DeriveLens selector callback.

• `rr.send_columns` hand-assembled from a custom parser → use the matching

reader; it produces chunks directly.

• Parsing MCAP/URDF with a non-Dalaran library then re-logging → McapReader

/ UrdfTree.

• `Chunk.from_columns` for data a reader already decodes (Pinhole

intrinsics, VideoStream, Transform3D from a transforms topic) → keep it in the reader stream; fix with a MutateLens if needed.

A wall of pyarrow.compute "missing-attribute" type errors (pc.filter, pc.list_element) usually means pc.* calls sit in module-level helpers instead of inside Selector.pipe lens callbacks. Refactor into a lens before suppressing the checker — the errors are a smell that the hand-building should not exist.

Porting an existing converter? Hand-built converters predate decoder improvements and are not ground truth. Re-verify the decoder output (step 2) and check every Chunk.from_columns / for-loop against this list before copying.

Core model

• LazyChunkStream is a lazy pipeline DAG, not a collection. Building

filters, lenses, maps, splits, and merges reads no source data.

• Execution starts at terminal calls: write_dlr(...), collect(),

to_chunks(), or iterating the stream.

• Execution is streaming, multithreaded, and mostly GIL-free. Prefer

stream/lens operations and PyArrow compute over Python row loops.

• Move semantics: builder calls (filter, drop, lenses, map,

flat_map) consume the input stream; reusing a consumed stream raises. Reassign after each step. Terminal calls do not consume, but each terminal call re-executes the whole pipeline; collect() once if that is too costly.

• ChunkStore is materialized in memory (stream.collect(),

ChunkStore.from_chunks). LazyStore is manifest-indexed, loads chunks on demand (RrdReader(path).store(), catalog segment stores). Both have schema(), summary(), stream(), and write_dlr(...).

Stream composition

from dalaran.experimental import Chunk, LazyChunkStream, OptimizationProfile

• stream.filter(content=, has_timeline=, is_static=, components=) keeps the

matching portion of each chunk; stream.drop(...) is its complement, same keyword filters. content takes an entity-path glob or a list of them.

• stream.map(fn) applies Chunk -> Chunk; stream.flat_map(fn) applies

Chunk -> Iterable[Chunk]. Escape hatches for chunk-level Python logic; prefer lenses for columnar work.

• stream.split(content=, ...) returns (matching, non_matching); both

branches share the same upstream.

• LazyChunkStream.merge(*streams) fans in any number of sources.
• LazyChunkStream.from_iter(chunks) wraps hand-built chunks.

stream = source_stream()  # any importer skill
stream = stream.drop(content="/video_raw/**")
stream = stream.lenses(fix_lens, content="/cam/**", output_mode="forward_unmatched")
merged = LazyChunkStream.merge(stream, sidecar_stream)
merged.write_dlr(out_path, application_id="my_app", recording_id=recording_id)

Hand-built chunks — sidecar only

Use Chunk.from_columns ONLY for data no reader or lens can emit — JSON/CSV calibration, frame offsets, external metadata. If a reader (McapReader/UrdfTree/ParquetReader) decodes the topic or a lens can derive it, that is the idiomatic path; do not hand-assemble it here. In the robot_data_preprocessing example the _only_ hand-built chunk is the JSON offsets sidecar; the camera fix, FK→/tf, meshes, and recolor are all readers + lenses.

Chunk.from_columns(entity_path, indexes, columns) mirrors rr.send_columns(...) and accepts the same archetype .columns(...) helpers. Empty indexes means static.

chunk = Chunk.from_columns(
    "/tf_static/robot_offsets",
    indexes=[],  # static
    columns=rr.Transform3D.columns(
        translation=translations,
        quaternion=quaternions_xyzw,
        parent_frame=parents,
        child_frame=children,
    ),
)
sidecar_stream = LazyChunkStream.from_iter([chunk])

rr.AnyValues.columns(...) covers non-standard metadata fields. For inspection, a Chunk exposes entity_path, num_rows, is_static, timeline_names, to_record_batch(), and format() (human-readable table).

Lenses

Lenses reshape, fix, or derive components without iterating rows. Apply with stream.lenses(lenses, output_mode=..., content=...).

• MutateLens(component, selector, keep_row_ids=False) modifies an existing

component in place.

• DeriveLens(component, output_entity=None, scatter=False) creates new

columns, optionally at another entity. Chain .to_component(descriptor, selector) per output; .to_timeline(name, "sequence" | "duration_ns" | "timestamp_ns", selector) extracts a time column from the data itself. scatter=True explodes one input row into N output rows (one per list element).

• Scope with content= whenever the same component name exists under multiple

entities.

Output modes, and the default is `drop_unmatched`:

• drop_unmatched (default): only lens outputs survive. Right for derive-only

intermediate streams; silently discards everything else if applied broadly.

• forward_unmatched: lens outputs plus the original components no lens

consumed. Right for targeted fixes that preserve the rest of the stream.

• forward_all: lens outputs plus all originals, including consumed ones. Can

duplicate data.

In-place fix (keep Arrow type and length intact):

stream = stream.lenses(
    MutateLens(
        "Pinhole:resolution",
        Selector(".").pipe(
            lambda res: pa.array(
                [(h, w) for w, h in res.to_pylist()],
                type=res.type,
            )
        ),
    ),
    content=["/external/cam_low", "/external/cam_high"],
    output_mode="forward_unmatched",
)

Derive with unit conversion (PyArrow compute, no Python loop):

DeriveLens("schemas.proto.JointState:message", output_entity="/joints_deg/waist").to_component(
    rr.Scalars.descriptor_scalars(),
    Selector(".joint_positions").pipe(lambda arr: pc.multiply(pc.list_element(arr, 0), 180.0 / math.pi)),
)

Selector grammar

Selector("<query>") navigates nested Arrow data, jq-style:

• . current value; .field struct field
• [] iterate list elements; [N] index a list
• ? suppress errors / skip missing optionals; ! assert non-null
• | pipe one expression into another

.pipe(fn) chains a Python/PyArrow transform (or another Selector). .execute(array) runs it eagerly; .execute_per_row(array) guarantees the output row count matches the input (use inside lens callbacks that must stay row-aligned).

Writing RRDs

• stream.write_dlr(path, application_id=..., recording_id=...) executes and

writes in one streaming pass.

• stream.collect(optimize=OptimizationProfile.OBJECT_STORE).write_dlr(...)

materializes, optimizes chunk layout, then writes. Memory scales with the materialized chunks.

• Profiles: OBJECT_STORE (large chunks, for storage/query/catalog) and

LIVE (small chunks, low-latency viewer).

• Multiple physical RRDs form one logical recording when they share a

recording_id; use this to separate base data, model/URDF data, and layers.

Always use `OptimizationProfile.OBJECT_STORE` when the DLR is headed for a Dalaran catalog or Hub, unless explicitly asked otherwise.

Chunk API vs logging API