Skip to content

6. Attach derived keys

Runnable companion: examples/03_attach_keys.py

An attach writes new keys onto samples that already exist: features, embeddings, predictions, metrics, preprocessed audio. It adds columns where an import or an add adds rows.

Attaching is the act; a producer is the identity performing it. Pick a producer id that names the thing that computed the values — f0_v001, whisper_large_v3, beam4 — because that id is how the values are found and compared later.

The loop

Three steps, and no step needs a list of sample ids:

from datavo_sdk import (
    SampleStream, HasKeys, AttachedBy, Not, SourceSpec,
    ShardPlan, TarWriter,
)

# 1. select the work: samples with the input, missing this producer's output
workset = client.create_workset(
    sources=[SourceSpec(
        sample_store="voice_corpus_v1",
        keys=["audio.wav"],
        filters=[
            HasKeys(["audio.wav"]),
            Not(AttachedBy("f0_v001", ["f0.npy"])),
        ],
    )],
    shard_plan=ShardPlan(shard_size=256),
)
if not workset.sample_count:
    return                                    # nothing to compute

# 2. stream the inputs and compute
with TarWriter("f0_shard.tar") as writer:
    for sample in SampleStream.from_workset(
        workset.workset_id, keys=["audio.wav"], client=client, decode=False
    ):
        f0 = compute_f0(sample["audio.wav"])
        writer.write({"__key__": sample["__key__"], "f0.npy": f0})

# 3. attach the result
result = client.attach_keys(
    "voice_corpus_v1",
    "f0_shard.tar",
    keys=["f0.npy"],
    producer_id="f0_v001",
    metadata={"producer_kind": "f0", "fingerprint": "f0@1.2.0"},
)
result.state              # "committed"
result.sample_count       # samples whose revisions were finalized
result.updated_sample_ids # the canonical ids that got the new revision

Selecting the work

An attach workset is an ordinary workset (chapter 8) with one predicate that subtracts what is already attached. A row filter is a flat list of predicates, each contributing a set of samples:

  • HasKeys([...]) — keep samples that have the inputs you need to compute.
  • Not(AttachedBy(producer_id, output_keys)) — drop samples this producer has already written those keys for. AttachedBy names the coverage; Not subtracts it, and it wraps any predicate (chapter 8). The first time the producer runs it has attached nothing, so nothing is subtracted and the workset is every sample with the inputs.

selectors= name the keys to read, and shard_plan sets how many samples travel per shard.

This is convergence — the second of datavo's two idempotency mechanisms, and the reason to select work rather than enumerate it. Two properties follow:

  • Doneness is an empty workset. sample_count == 0 means there is nothing to compute. A re-run after a successful pass is a no-op, not a recompute.
  • Partial progress resumes. An attach that covered half the store leaves a workset containing exactly the other half.

The boundary is (producer_id, key). A second producer writing the same key sees every sample as its own work — which is what makes two producers comparable rather than competing; the same producer sees none.

Computing

SampleStream.from_workset(..., decode=False) streams the selected samples in process: each one is a {"__key__": sample_id, "audio.wav": b"..."} mapping. decode=False is the producer setting — you forward payloads rather than interpret them. (Chapter 8 covers decoding, prefetch and the full read model.)

Write outputs with TarWriter, keyed by the canonical sample_id you were handed. writer.write({"__key__": sample_id, "f0.npy": array}) emits a member named <sample_id>.f0.npy, which is exactly what the attach expects — the mirror image of how you read.

Attaching

attach_keys is reserve → upload → commit → poll to a terminal state, in one call. The commit is worker-deferred, so it does not trust the commit response: it polls the ingest and returns the server's authoritative state. A commit whose HTTP response is lost to a timeout degrades to the same poll, which makes the call safe to retry.

The first attach also creates the producer, recording your metadata. Nothing else is required to start attaching: a producer is a caller-owned id, plus an optional display name and metadata — the same three things you own on a store. It has no lifecycle to manage and no version. Its output keys are datavo's bookkeeping: every attach adds the keys it wrote, so a producer that later attaches a second key is recorded as writing both.

Producers are scoped to a store, so f0_v001 in two stores is two producers.

Attaching twice

An attach also carries the operation_id from chapter 4, and the two mechanisms do different jobs:

  • replay (operation_id) — the same id returns the ingest already opened for it, so a retried job uploads once. Omit it and the SDK mints a fresh one.
  • convergence (the workset above) — a re-run computes only what is missing.

Replay protects the operation; convergence protects the work. Neither replaces the other, and idempotency has the full rules.

Register explicitly only when you want the declaration to exist before the first attach:

client.register_producer(
    "voice_corpus_v1", "f0_v001",
    declared_output_keys=["f0.npy"],
    metadata={"producer_kind": "inference", "maker": "pitch-extractor",
              "fingerprint": "f0@1.2.0", "run_ref": "wandb://…"},
)

Provenance is caller-owned metadata: a producer is generic, so there are no dedicated columns for run refs, checkpoints or configs. Put what you will want to search for in there. client.list_producers(store) lists what a store has.

Draining a big pass

A workset is live, and an attach workset excludes what is already attached, so the loop drains itself. Claim, attach, then reopen the claim surface for another round:

while True:
    ws = make_attach_workset(client)          # the create_workset call above
    if not ws.sample_count:
        break
    produce_and_attach(ws)
    client.reopen_workset(ws.workset_id)

Nothing needs extending or refreshing: the workset lives as long as it is being used (claiming counts) and is released once it goes idle.

How far along am I?

cov = client.producer_coverage("voice_corpus_v1", "f0_v001")
cov["total_samples"], cov["complete_samples"], cov["missing_samples"]
cov["completion_ratio"]   # 0.0 – 1.0
cov["state"]              # "empty" | "partial" | "complete"

Coverage is computed set-wise rather than per sample, so it is cheap enough to poll from a progress display.

Pitfalls

  • Writing against raw identity. Members must be keyed by the canonical sample_id Datavo gave you, never a source filename or the tar __key__ you imported with. Raw-identity attaches do not resolve.
  • Attaching keys the samples do not exist for. An attach only adds keys to existing samples. Bringing new samples in is an import or an add (chapters 4–5).
  • A new checkpoint under the same producer id. Convergence will skip everything that producer already wrote. New values mean a new producer id.
  • Expecting a "set tag" call. Producers do not write tags. Attach a scalar key and it becomes a searchable attribute on its own.

Next: 7. Store operations — read the graph you have been building.