Metadata-Version: 2.4
Name: dataio-rs
Version: 0.4.10
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: numpy>=2.0,<3 ; extra == 'bench'
Requires-Dist: pillow>=10,<12 ; extra == 'bench'
Requires-Dist: torch>=2.6,<2.10 ; extra == 'bench'
Requires-Dist: torchvision>=0.21,<0.25 ; extra == 'bench'
Requires-Dist: opencv-python-headless>=4.10,<5 ; extra == 'bench'
Requires-Dist: safetensors>=0.4.5,<0.7 ; extra == 'bench'
Requires-Dist: dataio-rs[bench] ; extra == 'bench-dali'
Requires-Dist: nvidia-dali-cuda120>=1.44,<1.52 ; extra == 'bench-dali'
Requires-Dist: dataio-rs[bench] ; extra == 'bench-spdl'
Requires-Dist: spdl>=0.0.8,<0.1 ; python_full_version >= '3.10' and extra == 'bench-spdl'
Requires-Dist: dataio-rs[bench] ; extra == 'bench-tensorflow'
Requires-Dist: tensorflow-cpu>=2.18,<2.21 ; extra == 'bench-tensorflow'
Provides-Extra: bench
Provides-Extra: bench-dali
Provides-Extra: bench-spdl
Provides-Extra: bench-tensorflow
License-File: LICENSE
License-File: THIRD_PARTY_NOTICES
Summary: High-performance data processing library for ML workloads.
Home-Page: https://github.com/Mikubill/dataio-rs
License-Expression: Apache-2.0
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Package, https://pypi.org/project/dataio-rs/
Project-URL: Repository, https://github.com/Mikubill/dataio-rs

# dataio

[![CI](https://github.com/Mikubill/dataio-rs/actions/workflows/ci.yml/badge.svg)](https://github.com/Mikubill/dataio-rs/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/dataio-rs.svg)](https://pypi.org/project/dataio-rs/)
[![Python](https://img.shields.io/pypi/pyversions/dataio-rs.svg)](https://pypi.org/project/dataio-rs/)

A Rust-backed data-loading library for files, object stores, tensors, and
columnar data. The native runtime fetches, decodes, transforms, and batches
across a shared worker pool, with NumPy, PyTorch, DLPack, and Arrow outputs.
One URI surface covers local files, S3 / R2, GCS, Azure, and Hugging Face;
direct reads and reusable pipelines share the same IO layer.

```python
import dataio

pipe = dataio.pipeline(
    dataio.slot("image")
        .decode_image(mode="rgb")
        .resize_short(256)
        .center_crop(h=224, w=224)
        .normalize("imagenet"),
    dataio.batch(64, prefetch=4),
)

with pipe.bind(records).run("torch", concurrency=64) as loader:
    for batch in loader:
        train_step(batch["image"])
```

## Why dataio

- **Native fast path.** Decode, resize, normalize, and collate are Rust
  kernels on a shared pool. Native IO and kernels release the GIL; optional
  Python source and map callbacks retain their Python execution costs.
- **Shared-buffer outputs.** Native tensors can reach PyTorch through DLPack,
  and compatible reads can view memory-mapped storage. Copies depend on
  mutability, alignment, layout and device transfer; retained views own their backing storage.
- **One surface, every backend.** `s3://`, `gs://`, `az://`, `hf://`,
  and local paths share the same pipeline ops and the same flat IO verbs.
- **Reproducible streams.** Deterministic sharding, seeded shuffle, and
  checkpoint resume that replays a partial batch so transform RNG stays
  aligned across a restart.

## Install

```bash
pip install dataio-rs
```

Python 3.11 or newer. Wheels are CPU-safe by default; CUDA, nvJPEG, and
cuFile / GDS are discovered lazily when used, so the same
wheel runs on a laptop or an 8-GPU node. Print the capability report for
the current host:

```bash
python -m dataio
```

## Pipeline

A pipeline is a static plan: optional record maps, one or more slot
chains, exactly one shape stage (`batch` / `collate` / `group_by`), and
optional post-batch maps or native batch transforms. Building it does no
work — binding a record source and calling `.run()` starts the runtime.

```python
pipe = dataio.pipeline(
    dataio.map(lambda r: {"image": r["path"], "label": r["class_id"]}),

    dataio.slot("image")
        .decode_image(mode="rgb")
        .resize_short(256)
        .center_crop(h=224, w=224)
        .normalize("imagenet"),

    dataio.slot("label").passthrough(),
    dataio.batch(64, drop_last=True, prefetch=4),
)
```

Op factories are grouped into namespaces; the slot methods
(`.decode_image()`, `.resize()`, `.normalize()`, …) are shorthand for the
same native ops.

| Namespace | Purpose |
|---|---|
| `dataio.decoders` | image, audio, json, npy, safetensors, text, bytes, pt |
| `dataio.transforms` | resize, crop, normalize, augment, conditional ops |
| `dataio.batch_transforms` | mixup, cutmix, padding, shuffle, normalization |
| `dataio.cond` | predicates for slot-local conditional transforms and skips |

Audio decoding can opt into anti-aliased sample-rate conversion with
`slot("audio").decode_audio(sample_rate=16000, resampling="sinc")` or
`dataio.decoders.audio(sample_rate=16000, resampling="sinc")`. The default
`resampling="linear"` preserves existing behavior and costs less computation.

### Record sources and shuffle

Record sources compose before native decoding. `map_records(fn)` maps one
record to one record; `flat_map_records(fn)` emits zero or more child records
in order. `concat(...)` appends sources without opening later ones early:

```python
def expand(record):
    for index, payload in enumerate(record["parts"]):
        yield {"key": f"{record['key']}:{index}", "payload": payload}

records = dataio.records(group_a).concat(group_b).flat_map_records(expand)
```

Return an iterable of complete child records; wrap a single mapping or byte
record in a list. Expansion has unknown length and requires bounded shuffle.
Concatenation has a known length only when all children do; an infinite child
prevents later sources from being reached. See the runnable
[source composition example](examples/iterable_source.py).

Bind a reusable iterable or a zero-argument record factory for repeated passes.
`from_iterable(...)` is explicitly single-pass, even when given a factory;
construct another adapter for another pass. `bind` uses indexed access for lists,
tuples and other `collections.abc.Sequence` instances. Other objects use an
explicit `__iter__`, or Python's legacy `__getitem__` iteration when absent;
`__iter__ = None` disables iteration for non-Sequence objects. A zero-argument
callable can still be a `bind` factory. `from_iterable` follows Python's `iter()`
protocol when an iteration method is present, including rejection of
`__iter__ = None`; it does not switch such an object to factory mode.

For a finite third-party dataset, explicitly choose integer row access with
`dataio.records(dataset, indexed=True)`. This declares that integer
`__getitem__` returns one record and requires a finite `__len__`; it bypasses
the dataset's iterator without reading any rows at construction. It accepts
reusable datasets, not one-shot streams, mappings or individual paths/buffers.
An object with integer column lookup needs a row adapter first.

With `batch(N)`, indexed sources may implement `__getitems__(indices)` to read
the requested rows together. Return a **list or tuple of exactly that many
records**, in the requested order, including duplicates. The loader snapshots
the outer container's references before running row maps; it does not deep-copy
the records. The whole fetch completes before per-record maps run, so side-effect
ordering differs from alternating scalar reads and maps. Exceptions and invalid
results propagate without retrying scalar reads. An absent method or `None`
uses scalar access; other non-callable values are errors. This protocol does not
interpret `dataset[indices]` column dictionaries as rows.

Unbuffered indexed shuffle forwards remapped indices to the bulk method.
Filtering, expansion and concatenation do not forward the original bulk method.
`collate()` and `group_by()` continue to read items individually. See the
[indexed/callable example](examples/callable_source.py) for bulk reads followed
by a custom Python collator.

Record sources also support bounded shuffle:

```python
records = dataio.webdataset("/data/shard-{0000..0099}.tar").shuffle(
    seed=42, buffer_size=4096,
)
```

`buffer_size` bounds pending records in the shuffle buffer; it is not a byte
or total-memory limit. Buffered shuffle mixes nearby records rather than
producing a uniform permutation of the whole dataset. It also works with
single-pass `from_iterable` sources. A fixed seed repeats the order only when
the source is re-iterable and unchanged; use a different seed for each epoch
to reshuffle. Omitting `buffer_size` retains the full permutation behavior,
including materialization of non-indexed sources and rejection of single-pass
sequential sources. Mapping and filtering after a buffered shuffle stay lazy.

### Python transforms and structured batches

Source transforms and `dataio.map(...)` before the batch boundary see input
records, before native decoding. Native slot chains perform decoding and tensor
transforms. Python maps after `batch`, `collate` or `group_by` receive the
decoded `Batch` and may return any nested mapping, tuple or application object.
They run on the consumer thread; they are not callbacks between native ops.

For an existing Python `collate_fn`, pass raw samples through an object slot:

```python
pipe = dataio.pipeline(
    dataio.slot("sample").passthrough(),
    dataio.batch(32),
    dataio.map(lambda batch: collate_fn(batch["sample"])),
)
```

Here each input record contains a `sample` value. For native tensor slots, use a
batch map to assemble the desired nested result from `batch["slot_name"]`.
`collate()` itself preserves upstream batch boundaries; it does not take a
Python collate function. Native batch transforms must precede Python batch maps.

Padding and batch normalization retain row keys, indices, original image sizes
and passthrough values. MixUp and CutMix combine rows, so those metadata accesses
remain unavailable afterward; `len(batch)` and `repr(batch)` still work.

### Slot-local conditionals

`dataio.cond` builds native predicates that compile into the plan
signature. Field predicates run before decode, so they can gate optional
slots; shape predicates run after decode and live inside transform
branches.

```python
from dataio import cond as c

pipe = dataio.pipeline(
    dataio.slot("image").decode_image(mode="rgb"),
    dataio.slot("mask", optional=True)
        .active_when(c.col("mask_id").is_not_null())
        .decode_image(mode="gray"),
    dataio.batch(32),
)

# Shape-driven branching after decode
dataio.slot("image") \
    .decode_image(mode="rgb") \
    .when(c.shape(axis=0).gt(512)) \
    .then(dataio.transforms.center_crop(h=512, w=512)) \
    .otherwise(dataio.transforms.resize(h=512, w=512))
```

### Grouped batches

When an upstream sampler owns batch boundaries, `group_by` emits one
batch per contiguous run of a key. Pair it with `from_sampler` to
adapt a PyTorch-style `BatchSampler` (or any iterable yielding lists of
rows). A passthrough slot carries per-record metadata alongside decoded data:

```python
pipe = dataio.pipeline(
    dataio.slot("payload").decode_numpy(),
    dataio.slot("meta").passthrough(),
    dataio.group_by("batch_id", prefetch=4),
)

records = dataio.from_sampler(
    batch_sampler,
    lambda row: {
        "key": row["id"],
        "payload": row["path"],
        "meta": row,
    },
)

with pipe.bind(records).run("numpy", concurrency=32) as loader:
    for batch in loader:
        arrays = batch["payload"]
        rows = batch["meta"]  # aligned with surviving batch.keys
```

`from_sampler` injects the `batch_id` field for you. The `record_fn`
argument is optional — omit it when the sampler already yields valid
record mappings. The adapter shallow-copies each mapping before adding the
batch ID; nested values remain shared.

## Runtime

`pipe.bind(records).run(...)` returns a context-managed loader. Worker
pools are process-wide and set once via environment variables;
per-loader knobs control admission and prefetch:

| Knob | Meaning |
|---|---|
| `run(..., concurrency=N)` | per-shard stage admission window; total activity also depends on prefetch depth |
| `dataio.batch(..., prefetch=K)` | number of batches in flight |
| `run(..., max_prefetch_bytes=B)` | optional bound on logical bytes of native preallocated outputs waiting for delivery |
| `DATAIO_CPU_WORKERS` | size of the shared CPU decode/transform pool |
| `DATAIO_IO_WORKERS` | size of the shared async IO runtime |
| `DATAIO_OBJECT_SHARDS` | HTTP / object-store shard fanout |
| `DATAIO_S3_HTTP_VERSION` | `h1` or `h2` transport selection |
| `DATAIO_TCP_QUICKACK` | Linux small-response ACK tuning; defaults to `1`, set `0` before creating connections to disable |

`concurrency="auto"` (the default) is enough for local data and moderate
batches; raise it for remote object reads or high-latency sources. Output
preserves submit order by default — pass `order="completion"` to yield
each batch as soon as it finishes.

The byte limit requires a fixed native output layout and a direct output writer,
such as image decode → resize → CHW normalization. Unsupported variable or
object outputs and oversized batches fail before allocation. The limit covers
pending native outputs, including reordered results; it is not an RSS limit and
excludes decode temporaries, pool retention, collation and outputs held by callers.

With `.run("torch", device="cuda:0")`, pinned CPU outputs wait for the target
CUDA stream's copy to finish before their host storage can be reused. This
preserves values at the cost of asynchronous copy overlap and small-transfer
latency. Native CUDA outputs keep their stream-ordered DLPack handoff.

### Error handling

```python
loader = pipe.bind(records).run("torch", on_failure="skip")
```

| `on_failure` | Behavior |
|---|---|
| `"strict"` | raise on the first row error |
| `"skip"` | drop failed rows, raise only if a batch has no survivors |
| `0`, `1`, … | require at least that many survivors per batch |

Per-row errors land on `batch.errors`. For automatic logging, pass
`log_errors=True` (routes through the `dataio.rows` logger) or a callable
for custom handling:

```python
loader = pipe.bind(records).run("torch", log_errors=True)
loader = pipe.bind(records).run("torch", log_errors=my_logger.warning)
```

Errors while iterating or expanding records, and Python batch-map errors,
propagate directly. A Python batch map runs after the loader has consumed that
batch; if it raises, checkpoint progress is not rolled back. Retrying a callback
or making its external side effects transactional is the caller's responsibility.

### Distributed training and resume

```python
loader = pipe.bind(records).run(
    "torch",
    shard={"id": rank, "of": world_size, "pad": True},
    seed=42,
    concurrency=64,
)
```

`pad=True` keeps every rank at the same batch count; `seed` drives
shuffling and native batch-transform RNG. `state_dict()` /
`load_state_dict()` checkpoint and resume the loader — mid-batch resume
replays the partial batch from its first sample, so transform RNG stays
aligned across the restart.

Resume re-enumerates and skips the expanded source prefix; it is not a saved
source cursor, shuffle buffer or partially consumed child iterator. Python loader
checkpoints include a separate `record_source_signature` for tracked source
configuration: adapter mode, ordered source transforms, shuffle settings,
composition and declared `CustomOp` versions. This leaves the native schema 3
and existing pipeline signature unchanged. A present malformed or mismatched
source signature is rejected before native restore, even at an epoch boundary.
An older checkpoint without this field can resume a source with no tracked
configuration; for a configured source, it requires an integer `samples_consumed`
of zero.

Keep source contents and callback behavior unchanged. The guard does not inspect
raw data, ordinary closures, hidden state or RNG progress; identically described
opaque children can still be swapped without detection. Sources and callbacks
must replay consistently. Expansion happens before native sharding,
so each rank may enumerate upstream records and run expansion callbacks for
records it does not emit; composition does not push partitioning into storage.

Mid-epoch checkpoints must come from `order="submit"`. Completion-order delivery cannot
be reconstructed from a consumed-sample count, so restoring such a checkpoint
raises an error. State schema 3 records output order explicitly. Schema 2
checkpoints are accepted automatically at epoch boundaries; an older mid-epoch
checkpoint needs an explicit migration after verifying it used submit order.

## Direct IO

Flat verbs for scripts, evaluation, and one-shot reads:

```python
dataio.head("s3://bucket/file.bin")
chunk = dataio.read("s3://bucket/file.bin", offset=0, size=1 << 20)
etag  = dataio.write("s3://bucket/out.bin", chunk)

dataio.exists("s3://bucket/out.bin")
dataio.ls("s3://bucket/prefix/")
dataio.glob("s3://bucket/**/*.safetensors")
```

For large eager reads, fill a caller-owned writable buffer instead of
allocating a fresh Python `bytes` object:

```python
buf = bytearray(1 << 30)
n = dataio.fetch_into("s3://bucket/checkpoint.bin", buf)
view = memoryview(buf)[:n]
```

Composable handles for repeated operations on one source:

```python
src = dataio.reader("s3://bucket/shard.tar.gz")
payload = src.gunzip().tar_entry("0001.jpg").read_all()

with dataio.writer("s3://bucket/checkpoint.bin") as w:
    w.write_chunks(huge_generator())
```

`src.read_buffer()` returns a read-only buffer view that retains its native
storage, avoiding the final copy into Python `bytes`. Use `memoryview(...)` or
`numpy.frombuffer(...)` when the consumer accepts the buffer protocol.

For sequential processing, `dataio.stream(uri)` opens through the response
headers and reads the body on demand. Its byte, line and SSE iterators share one
cursor; `close()` cancels pending reads. Lines and SSE events have configurable
8 MiB limits. Only `read_all()` explicitly materializes the remaining response.
Streaming never retries a body after exposing it to the caller.

Custom TLS roots use `SSL_CERT_FILE`, with `AWS_CA_BUNDLE` as a fallback. A
configured PEM bundle replaces the built-in roots; invalid bundles fail clearly.

Set `HTTP_PROXY`, `HTTPS_PROXY`, or `ALL_PROXY` before the first network request
to use an HTTP proxy. HTTP requests use forwarding; HTTPS uses CONNECT with
normal origin certificate verification. Proxy URLs support Basic authentication
and must use `http://`; HTTPS and SOCKS proxy transports fail explicitly.
Lowercase variables take precedence. `NO_PROXY` / `no_proxy` accept domain
suffixes, IP addresses, CIDR networks, optional ports, and `*`. CGI environments
(`REQUEST_METHOD` present) ignore uppercase `HTTP_PROXY`.

`dataio.configure_transport("https", timeout_secs=...)` bounds each eager request attempt,
including redirects, response headers and body. Streaming instead applies the
body timeout while waiting for data, excluding caller processing time.

Large eager reads can configure range planning per scheme:

```python
dataio.configure_transport(
    "https", timeout_secs=30, readall_chunk_mib=32, readall_concurrency=8,
)
```

The range defaults remain 32 MiB and 8, subject to environment overrides.
`readall_chunk_mib` is a chunk-size floor for automatic partitioning;
`readall_concurrency=1` disables splitting. These settings govern each read,
not aggregate memory or concurrency across readers. Small reads remain single
requests, and splitting still requires a strong validator. Configuration
replaces that scheme's explicit options: pass options together to retain them.

Credentials and endpoints are configured once per scheme:

```python
dataio.configure_credentials("s3", method="default")  # platform chain
dataio.configure_credentials("r2", access_key_id="...", secret_access_key="...")
dataio.configure_endpoint("r2", "https://<account>.r2.cloudflarestorage.com")
```

`DATAIO_CACHE_DIR` enables the optional local byte cache. Object/HF cache entries
are partitioned by a hash of a supported explicit credential and endpoint
identity. Dynamic credential providers bypass disk caching; their in-process
parsed indexes follow client configuration changes. Cache hits do not revalidate
origin freshness or permissions: use immutable sources or disable the relevant
caches when reading mutable objects.

`method="default"` resolves through env vars, a shared profile, instance
metadata, or workload identity — whichever the platform exposes.

## Format helpers

```python
arr   = dataio.read_numpy("/data/x.npy")
image = dataio.read_image("s3://bucket/image.png", mode="rgb")
cols  = dataio.read_parquet("/data/shard.parquet", columns=["key", "height"])
rows  = (dataio.scan_parquet("/data/shard.parquet")
            .filter(lambda r: r["height"] > 512)
            .collect())

for example in dataio.iter_tfrecord("/data/train.tfrecord"):
    label = example["label"]
```

Feather / Arrow IPC scans can export Arrow data without creating Python
objects for individual cells:

```python
scan = dataio.scan_feather("/data/table.arrow").select(["key", "value"])
stream = scan.collect_arrow_stream()  # Arrow C stream protocol
table = scan.collect_arrow()          # requires optional pyarrow
```

These exports preserve Arrow types and decoded buffers, including projection
and row slicing. `collect_arrow_reader()` returns a PyArrow batch reader;
Python callable filters use the row-oriented collection methods instead.
Native NPY decoding supports `complex64` and `complex128`, including foreign
byte order and Fortran layout.

HDF5 reads support integer indices, positive-step slices, and one ellipsis:

```python
window = dataio.read_hdf5(
    "/data/arrays.h5", key="images", selection=(slice(10, 20), ...),
)
with dataio.open_hdf5("/data/arrays.h5") as source:
    view = source.read(key="labels", copy=False)
```

The default returns independent writable NumPy arrays. `copy=False` returns
read-only output that remains valid after session close; decoding, selection,
or byte-order conversion can still allocate. Contiguous compatible datasets
can share the file mapping, while chunked selections decode only intersecting
chunks. Remote HDF5 still downloads the whole file when opened.

### safetensors

Zero-copy mmap reads for both single files and sharded checkpoints, including
`C64` / `complex64` tensors with NumPy, PyTorch, and DLPack outputs:

NumPy views are read-only. Torch uses private copy-on-write local mappings;
writable outputs from immutable in-memory or remote buffers require a copy.

```python
dataio.write_safetensors(
    "/fsx/ckpt-step-12000/",
    state_dict,
    max_shard_size="5GB",
    metadata={"format": "pt"},
)

state_dict = dataio.read_safetensors("/fsx/ckpt-step-12000/", framework="torch")
subset = dataio.read_safetensors(
    "/fsx/ckpt-step-12000/",
    keys=["model.layers.0.weight", "model.norm.weight"],
    framework="torch",
)

with dataio.open_safetensors("/fsx/model.safetensors") as st:
    tensor = st.read(key="layer.weight", framework="torch")
    subset = st.read(keys=["layer.weight", "layer.bias"], framework="torch")
```

On Linux hosts with CUDA and a GDS-ready filesystem, reads land directly
in device memory:

```python
state_dict = dataio.read_safetensors(
    "/fsx/ckpt-step-12000/",
    framework="torch",
    device="cuda:0",
)
```

`dataio.gds_available()` and `dataio.gds_info()` report the active state.

## RDMA transport

`dataio.rdma` is a native one-sided RDMA client + server over libfabric
(OFI), built for fast peer-to-peer checkpoint transport over EFA / verbs
/ RoCE. It is **CPU-safe**: `libfabric.so.1` is discovered via `dlopen`
at first use. `available()` reports whether a compatible RMA provider can be
opened, including `tcp;ofi_rxm` / `sockets` functional fallbacks; it does not
prove physical RDMA availability. `info()` reports the compiled feature and
discovered providers. Handles are context managers.

```python
import dataio.rdma as rdma

with rdma.serve(bind="0.0.0.0:0", buffer_size=1 << 20) as server:
    addr = server.local_addr
    with rdma.connect(addr) as client:
        rdma.write(client, offset=0, data=b"hello rdma")
        assert rdma.read(client, offset=0, size=10) == b"hello rdma"
```

The data path is **pipelined and zero-copy**: a transfer is split into
`chunk_size` segments with up to `max_inflight` posted concurrently
(the useful window depends on provider, payload and network). `read_into` lands bytes
straight into a caller-owned buffer (`bytearray` / `numpy` / pinned
`torch` tensor / `mmap`) with no intermediate copy; `write` registers the
caller's buffer directly as the RMA source.

Serving a checkpoint file needs no extra protocol — a self-describing
format like safetensors works as-is, and a CUDA-connected client can pull
each tensor's byte range straight into VRAM via **GPUDirect RDMA**:

```python
# producer: serve weights read-only
server = rdma.serve_file("/fsx/model.safetensors", bind="0.0.0.0:0")

# consumer: GPUDirect-RDMA tensor ranges into VRAM
client = rdma.connect(server.local_addr, cuda=True)
import torch
dst = torch.empty(shape, dtype=torch.bfloat16, device="cuda:0")
rdma.read_into(client, dst, offset=tensor_byte_offset)  # NIC → VRAM, host untouched
```

GPUDirect needs an HMEM-capable provider and libfabric API ≥ 1.9. Rather
than write a C `fi_getinfo` probe, run the self-diagnostic — it runs the
real plain-RMA and HMEM opens and reports exactly why each would or
wouldn't match:

```python
for line in rdma.diagnose()["summary"]:
    print(line)
```

`rdma.available()` / `rdma.info()` give the boolean and per-provider
capability snapshot.

## API at a glance

| API | Purpose |
|---|---|
| `pipeline`, `slot`, `slots`, `chain`, `batch`, `collate`, `group_by` | pipeline DSL |
| `map`, `custom_op` | Python extension hooks |
| `DataLoader` | synchronous batch iterator |
| `read` / `fetch_into` / `write` / `head` / `stream` / `ls` / `glob` / `exists` / `delete` | URI bytes plus explicit `read(format=...)` materialization |
| `reader`, `writer`, `scan` | composable IO handles and generic lazy scans |
| `from_callable`, `from_iterable`, `from_iterator`, `from_sampler`, `files`, `urls`, `manifest`, `webdataset`, `archive` | record sources |
| `read_safetensors`, `write_safetensors`, `open_safetensors` | safetensors |
| `read_numpy`, `scan_numpy`, `read_npz`, `scan_npz`, `read_image`, `read_parquet`, `scan_parquet`, `read_feather`, `scan_feather`, `iter_tfrecord`, `scan_tfrecord`, `read_hdf5`, `scan_hdf5`, `open_hdf5` | format helpers |
| `configure_credentials`, `configure_endpoint`, `configure_transport` | auth and transport |
| `gds_open`, `gds_register_buffer`, `gds_read_into_tensor` | low-level GDS |
| `rdma.serve`, `rdma.serve_file`, `rdma.connect`, `rdma.read_into`, `rdma.diagnose` | RDMA transport |

## Documentation

- [Architecture](docs/arch.md) — module responsibilities, runtime IR,
  buffer ownership, RDMA and security.
- [Benchmarking](docs/benchmark.md) — runnable commands, comparable workloads
  and regression gates.
- [Roadmap](docs/roadmap.md) — remaining gaps and acceptance criteria.
- [`examples/`](examples/) — runnable end-to-end scripts. Good starting
  points: [`examples/multi_slot_pipeline.py`](examples/multi_slot_pipeline.py),
  [`examples/safetensors_latents.py`](examples/safetensors_latents.py),
  [`examples/conditional_slots.py`](examples/conditional_slots.py).

## Development

The Rust toolchain is pinned in `rust-toolchain.toml`. Editable build:

```bash
uv run --no-sync --with maturin maturin develop --profile staging
```

Source builds omit the optional native RDMA shim and need no libfabric headers.
Add `--features rdma` to the maturin command to enable it; install the libfabric
development headers first. Published wheels include this feature and discover
the runtime provider when used. The native Rust I/O, object and format crates
can be built without Python; only the root Python adapter enables the runtime's
`python` feature.

Tests:

```bash
cargo test
uv run --no-sync python -m pytest python/tests -q
```

Wheel:

```bash
uv run --no-sync --with maturin maturin build --profile staging --out dist
```

## Performance

Measure the workload and output type used by your application. Decode-heavy
pipelines, object-only records, remote transfers and GPU operations have different
bottlenecks. The [benchmark guide](docs/benchmark.md) provides reproducible
commands, output checks and paired regression comparisons; no single workload
establishes performance across all backends.

## License

Apache-2.0. Source: <https://github.com/Mikubill/dataio-rs>. PyPI:
<https://pypi.org/project/dataio-rs/>.

