8. Read samples¶

Runnable companion: examples/05_read_samples.py

Reading is two calls: select a workset, then iterate it. Nothing has to be frozen, named, or planned first.

from datavo_sdk import SampleStream, SourceSpec, KeyGroupSpec, ShardPlan

workset = client.create_workset(
    sources=[SourceSpec(sample_store="voice_corpus_v1",
                        keys=["audio.wav", "transcript.txt"])],
    key_groups=[KeyGroupSpec(name="audio", keys=["audio.wav"]),
                KeyGroupSpec(name="text", keys=["transcript.txt"])],
    shard_plan=ShardPlan(shard_size=256),
)

for sample in SampleStream.from_workset(
    workset.workset_id, keys=["audio.wav", "transcript.txt"], client=client
):
    sample_id = sample["__key__"]           # canonical sample_id
    audio = sample["audio.wav"]             # bytes
    transcript = sample["transcript.txt"]   # str

A workset stores the selection, so creating one costs nothing and no bytes move until you iterate. It is released once you stop using it — there is nothing to clean up, and with client.workset(request) as ws: hands it back sooner if you want to be explicit.

What goes into a workset¶

Five parts, each answering a separate question:

Part	Question	Shape
`sources`	which samples, from where	one `SourceSpec` per store — the only answer to "where from" (chapter 10 unions several)
`filter` (per source)	which rows	a flat list of predicates
`keys` (per source)	which columns	key names, plus `producer_pins` when a key has competing producers (chapter 9)
`key_groups`	how columns travel	named groups; the unit of download and caching. Omit it and one `all` group over the sources' keys is derived
`shard_plan`	how many samples per shard	`ShardPlan(shard_size=…, seed=…)`

sources says where the samples come from, and a single store is the one-element case:

workset = client.create_workset(
    sources=[SourceSpec(sample_store="voice_corpus_v1", keys=["audio.wav"])],
    shard_plan=ShardPlan(shard_size=256),
)

key_groups is omitted here, so one all group over the source's keys is derived.

Row filters¶

A row filter is a flat list of predicates, ANDed together. Each predicate is a class, and its class is what it does:

from datavo_sdk import Attr, AttrIn, AttrRange, HasKeys, AnyKey, AttachedBy, Not

filters = [
    Attr("language", "en"),                          # attribute equals
    Attr("speaker", AttrIn(values=["s1", "s2"])),      # attribute in a set
    Attr("duration", AttrRange(gte=1.0, lt=30.0)),    # numeric range
    HasKeys(["audio.wav", "transcript.txt"]),         # has all of these keys
    AnyKey(["audio.wav", "audio.flac"]),              # has at least one of these
    AttachedBy("f0_v001", ["f0.npy"]),                 # this producer has covered it
]

Attributes are what chapter 4 introduced: import-time default_tags plus any key whose payload decodes to a scalar. client.list_store_attribute_names(store) and list_store_attribute_values(store, name) tell you what you can filter on.

The order you write them in is your convenience; the server compiles them in a canonical order, so two equivalent filters are the same workset.

Negation¶

Not(...) wraps any of them and selects what that predicate does not:

filters = [
    HasKeys(["audio.wav"]),
    Not(HasKeys(["qc.reject"])),                      # no qc.reject key
    Not(AnyKey(["audio.flac", "audio.mp3"])),          # neither of these keys
    Not(Attr("language", "en")),                      # language is not en
    Not(AttachedBy("f0_v001", ["f0.npy"])),            # f0_v001 has not covered it
]

Two things to know about it:

Absence matches. A negation is the complement of a set, so it selects samples that carry no such attribute or key at all. Not(Attr("language", "en")) includes samples with no language attribute. This is the same reason Not(AttachedBy(...)) selects every sample the first time a producer runs: it has attached nothing, so nothing is subtracted. To say "has a language, and it isn't en", name the values you do want: Attr("language", AttrIn(values=["de", "fr"])).

The conjunction is negated as a whole. Not(HasKeys(["a", "b"])) is "lacks a, or lacks b" — the complement of having both. "Has neither" is Not(AnyKey(["a", "b"])).

Predicates are ANDed, and a negation is one more conjoined predicate — so a filter says "all of these, and none of those". There is no OR of negations: the one OR in the language is AnyKey, and De Morgan covers the useful case, since Not(AnyKey([a, b])) is exactly "neither a nor b".

Key groups¶

A key group is a named bundle of keys — inputs, features, labels. Groups are the unit of shard download and cache granularity: for each shard, one tar per group you touched, merged by sample_id as you iterate. Two readers asking for overlapping key sets share the group tars they have in common, even when their requested key lists differ.

Every key belongs to a group, so grouping keys that are read together is the optimization that matters: a run reading only inputs never downloads features.

Iterating¶

SampleStream.from_workset(...) yields one dict per sample, __key__ plus the keys you asked for, auto-decoded by extension:

extension	you get
`.txt`	`str`
`.json`	`dict` / `list`
`.cls`	`int`
`.npy`, `.npz`	NumPy array
`.pth`	result of `torch.load`
anything else	raw `bytes`

Pass decode=False to get raw bytes throughout — the right setting when you are forwarding payloads rather than interpreting them (chapter 6).

keys= is required when reading a workset: you say what you want, and the workset has to carry it.

Prefetch. Up to eight shard downloads stay in flight by default so transfer overlaps decode. prefetch=N tunes it; prefetch=0 is strictly sequential.

Caching. Shards are content-addressed, so a second pass — same process, a new process, or another machine sharing a cache tier — reads from local disk without touching the server. See the shard cache.

PyTorch. SampleStream has no torch dependency. For DataLoader integration with per-worker and per-rank shard partitioning, use the subclass:

from datavo_sdk.torch_dataset import DatavoTorchDataset
from torch.utils.data import DataLoader

ds = DatavoTorchDataset.from_workset(workset.workset_id, keys=["audio.wav"])
loader = DataLoader(ds, num_workers=4, batch_size=8)

Importing datavo_sdk.torch_dataset without torch installed raises ImportError rather than degrading silently, and .pth decoding needs torch too.

Live means live¶

A workset answers as of now. That suits exploring, attaching and debugging, and it is wrong for anything that must be reproduced: the same workset answers differently once the store grows. A stable source is a converted one — chapter 11.

Pitfalls¶

Asking for a key the workset does not carry. Requested keys must be in the workset's key groups; add them to the source's keys first.
Iterating twice and expecting identical rows. Two passes over a live workset can differ if the store changed in between.
A client that goes out of scope. Iteration is lazy and streams through the client. Keep it alive for the whole loop.
Tiny shards. shard_size is the download unit. Single-sample shards maximize per-shard overhead; a few hundred is a normal starting point.

Next: 9. Consume derived keys — read what a producer wrote.