batchcorder

A Rust-backed Python library for caching Arrow record-batch streams so they can be replayed multiple times from a source that can only be read once.

The problem

Arrow RecordBatchReader is a single-use stream — once consumed, it is gone. Training loops and multi-pass data pipelines need to iterate the same stream repeatedly without re-reading from disk or the network each time.

What batchcorder does

StreamCache wraps any Arrow stream source (anything that implements __arrow_c_stream__) and stores each RecordBatch in a cache. Two storage modes are supported:

• Memory-only (default): batches are stored as reference-counted pointers in RAM; reads are zero-copy and no IPC serialisation happens.
• Hybrid memory+disk: batches are serialised to an append-only IPC file on disk; a configurable hot layer keeps recently ingested batches in RAM to reduce disk I/O.

Multiple independent readers can replay the stream concurrently, each maintaining their own position in the batch sequence.

flowchart LR
    U["upstream source<br/>(read once)"] --> D["StreamCache<br/>[mem cache / mem+disk cache]"]
    D --> R0["StreamCacheReader 0<br/>(from batch 0)"]
    D --> R1["StreamCacheReader 1<br/>(from batch 0)"]
    D --> R2["StreamCacheReader 2<br/>(from batch 3)"]

Installation

pip install batchcorder

Usage

import tempfile

import pyarrow as pa

from batchcorder import StreamCache

table = pa.table({"x": [1, 2, 3], "y": [4, 5, 6]})

# Memory-only (default capacity = total physical RAM)
ds = StreamCache(table.to_reader(max_chunksize=1))

# Memory-only with explicit capacity
ds = StreamCache(
    table.to_reader(max_chunksize=1),
    memory_capacity=64 * 1024 * 1024,  # 64 MB
)

# Hybrid memory+disk
tmp = tempfile.mkdtemp()
ds = StreamCache(
    table.to_reader(max_chunksize=1),
    memory_capacity=64 * 1024 * 1024,  # 64 MB in RAM
    disk_path=tmp,
    disk_capacity=512 * 1024 * 1024,  # 512 MB on disk
)

# Replay as many times as needed
for batch in ds:
    print(batch)

# Or get an independent reader handle
reader = ds.reader()
result = pa.RecordBatchReader.from_stream(reader).read_all()

# Bounded-memory: evict batches once all readers have passed them
ds = StreamCache(
    table.to_reader(max_chunksize=1),
    max_readers=2,  # at most 2 reads; batches evicted when both advance
)

# Pre-ingest everything upfront
ds.ingest_all()

Compatibility

StreamCache and StreamCacheReader implement both __arrow_c_stream__ and __arrow_c_schema__, so they work with any Arrow-compatible library:

import pyarrow as pa
import duckdb

pa.table(ds)  # PyArrow
pa.table(ds.reader())  # via StreamCacheReader
duckdb.table("ds")  # DuckDB

Key properties

• Single-read source: the upstream stream is consumed exactly once; all subsequent reads come from the cache.
• Concurrent readers: multiple StreamCacheReader instances from the same stream cache are fully independent and thread-safe.
• Lazy ingestion: batches are fetched from the upstream source on demand as readers advance, not upfront.
• Replay from any position: ds.reader(from_start=True) (default) replays from batch 0; ds.reader(from_start=False) starts from the current ingestion frontier (next batch not yet ingested).
• Bounded-memory streaming: set max_readers=N to evict batches once all N readers exist and have advanced past them — eviction does not begin until all N readers have been created. max_readers is a hard cap on total readers ever created (dropping a reader does not free a slot). Once eviction has started, reader(from_start=True) raises ValueError. When unset, all batches are retained indefinitely.

Development

# Install dependencies and build the extension
uv sync --no-install-project --dev
uv run maturin develop --uv

# Run tests
uv run pytest

# Run all pre-commit checks
uv run pre-commit run --all-files