docs(interleaved): add image payload cache tutorial

[VibhuJawa]

Why

Interleaved corpora store images by reference, and the same image is referenced by many documents. Materializing a corpus therefore fetches the same object repeatedly.

Measured on MINT-1T HTML:

Quantity	Value
Image occurrences	1,582,193,028
Unique images	356.0M
Mean references per image	4.445
Images with exactly one reference	40.2%
Highest observed reference count	1,584

Every unique image must be fetched at least once, so the ceiling on avoided fetches is 1 - 356.0M / 1,582,193,028 = 77.5%. The distribution is skewed: 40.2% of images are referenced once and can never produce a hit, while a long tail referenced hundreds of times produces almost all of the savings.

materialize_task_binary_content already deduplicates identical byte ranges within a task, but references to one image are spread across thousands of tasks and many workers, so the repeats survive.

Cache design

PayloadCache is a content-addressed directory keyed by the source_ref JSON locator. A hit returns the payload without touching the source; a miss falls through to normal I/O and stores the result.

• Keys are hashed into a two-level fan-out, so no directory holds millions of entries.
• Writes land in a sibling temp file and are renamed into place, so a reader never sees a partial payload and concurrent writers of the same immutable payload are harmless.
• Faults are never fatal: a failed read is a miss, a failed write is skipped.

Two preconditions the README now states up front

The key is the locator, not the image. The cache pays off exactly as often as the same source_ref recurs. That holds for a deduplicated store that keeps one copy per unique image and points many documents at it. It does not hold for a WebDataset-style layout that writes a separate copy into every sample's shard — no locator repeats and the hit rate is zero however often the image itself recurs.

The source must be slower than the cache. Measured at a 140 kB mean payload on the shared filesystem (WekaFS), 16 workers: a put costs 10.5 ms and a hit get 2.0 ms. Against a ~0.5 s object-store GET a hit is ~250x cheaper. Against a byte-range read into a local packed tar — 1.3 ms for the same payload — the cache is a net loss, because it replaces a few large batched range reads with many small whole-file reads. Cache a remote source; do not cache a fast local one.

The measured concurrency caveat

4.4x is a serial limit, and a single concurrent pass gets a fraction of it. Measured end to end on a corpus whose static-oracle ideal hit rate was 0.4720:

Workers	Hit rate	Source reads
1	0.4720 — exactly the ideal	18,553
16	0.0906	32,993

This is not a defect in the cache: at one worker the measured hit rate matched the oracle to the last digit. A worker that touches a key for the first time simply cannot see a write an in-flight peer has not made yet, so the single-pass hit rate is governed by partitions / workers rather than by reference multiplicity.

What recovers it, in the README:

• Key-affinity routing — send every occurrence of a key to the same worker, so repeats serialize behind the first fetch. This is what restores the full multiplicity in one pass.
• More passes. A second run over the same shards hit 100% with zero source reads, so multiple epochs, retries and repeat jobs realise the full reference count.
• Many more partitions than workers.

Sizing

Static-oracle hit rate for a full pass, exact per-image sizes, greedy admission by reference count:

Cache size	Hit rate
1 TB	20.2%
2 TB	31.0%
3 TB	38.6%
5 TB	48.9%
Unbounded (51 TB)	77.5%

Returns are sublinear, so size the cache against the cost of the fetches it removes, not against the corpus. Admitting by value density (k-1)/size instead of by reference count raises the 2 TB figure to 46.4% — worth roughly a 2.2x larger cache. PayloadCache implements neither admission nor eviction today: it stores every miss, so bound it with a filesystem quota and clear it between corpora.

Contents

tutorials/interleaved/image_payload_cache/ — a main.py following the existing interleaved tutorial conventions (RayClient, Pipeline, RayDataExecutor) whose only tutorial-local code is a CachedMaterializeStage that hands a PayloadCache to materialize_task_binary_content, plus a README covering the preconditions, the reference-count motivation, the concurrency caveat, sizing, and run instructions.

The tutorial keeps its own materialize stage on purpose: shipped filter stages take payload_cache= directly, but they materialize into a scratch task to compute their keep-mask and do not emit the bytes, so a pipeline that must both filter and write payloads still needs a materialize stage of its own.

Verification

Run end to end on a synthetic four-partition corpus (8 unique images, 24 references, real tar member offsets): the cold pass wrote exactly 8 cache entries and materialized all 24 image rows with zero errors, and a second pass materialized all 24 rows with zero errors after the source tar was removed — every payload came from the cache.

ruff check and ruff format --check pass on the tutorial. pytest tests/stages/interleaved -m "not gpu" is 291 passed / 6 skipped / 4 failed, where the 4 failures are pre-existing loguru-to-caplog propagation failures in test_multimodal_reader.py and test_schema_utils.py, in files this branch does not touch.

🤖 Generated with Claude Code