Metadata-Version: 2.4
Name: batchcorder
Version: 0.1.3
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: pyarrow>=18
License-File: LICENSE
Summary: A Rust-backed Python library for caching Arrow record-batch streams so they can be replayed multiple times
Author: Daniel Mesejo
Requires-Python: >=3.10
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Documentation, https://batchcorder.readthedocs.io
Project-URL: Homepage, https://github.com/mesejo/batchcorder
Project-URL: Repository, https://github.com/mesejo/batchcorder

# 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.

```mermaid
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

```bash
pip install batchcorder
```

## Usage

```python
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:

<!-- skip: next -->
```python
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

```bash
# 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
```

