rypipe

Format-agnostic columnar ingestion engine. Rust core, Python bindings.

Parse row-oriented byte streams into Apache Arrow tables with parallel scheduling, memory-bounded execution, query pushdown, and a chainable pipeline API. Format adapters live in separate packages.

---

What is rypipe

rypipe is a format- and source-agnostic ingestion framework that provides a common execution runtime for turning arbitrary record-oriented data sources into typed columnar data. It separates format-specific parsing from format-agnostic execution, so the same engine can parse XML, JSON, CSV, HTML, or any other row-oriented format once you provide a small adapter.

Add a new format by implementing two small traits: Splitter and RecordParser.

rypipe was originally developed as the ingestion engine for crxml, a Crystal Reports XML parser, and was later extracted and abstracted. The engine's design, performance characteristics, and API were shaped by real-world production use with crxml. We use crxml as the primary example throughout the documentation because it demonstrates the full power of the framework: complex nested schemas, large files, parallel processing, and advanced filtering.

Quick start

Any parser you build on rypipe (or any adapter package you install) gets this same API: a small Rust crate behind a clean, chainable Python interface with fused filtering, parallel parsing, and bounded-memory streaming. The quick start below shows crxml as a concrete example; your format's adapter would look and behave the same.

pip install crxml

import crxml
from crxml import CrystalXMLSource, CastTypes, FilterRows, col

source = CrystalXMLSource("report.xml", row_tag="Details")

# Simple read
table = source.to_arrow()

# Pipeline with stages
result = (
    source
    | CastTypes({"Amount": float})
    | FilterRows(field="Status", op="==", value="Active")
    | crxml.to_arrow()
)

# Expression predicates fuse into the Rust parse loop too
result = (
    source
    | FilterRows((col("Amount") > 100) & (col("Status") == "Active"))
    | crxml.to_arrow()
)

# Convert to DataFrame
df = source.to_pandas()

# Constant-memory streaming
for batch in source.iter_record_batches(memory="64MiB"):
    writer.write_batch(batch)

Why rypipe

• One runtime, many formats. XML, JSON, CSV, HTML, TSV, and any future format share the same parallel scheduler, memory-bounded executor, and pushdown infrastructure. An adapter is two small traits, not a full engine. crxml (Crystal Reports XML) is the reference adapter that proved this model.

• Measured performance. On the documented 533 MB crxml workload, ~4.2 GB/s parallel and ~950 MB/s single-threaded (Ryzen 7 5800X). Arrow export moves string/dictionary buffers without copying; primitive arrays are copied. Predicate-first evaluation. Layout prediction via memcmp.

• Correctness by construction. Differential testing, fuzz targets, property tests, and a tier-ladder profiler.

• Python-native ergonomics. Chainable pipeline API with automatic fusion of rename/drop/cast/filter into the Rust parse loop. Streaming with bounded memory. Schema discovery. DataFrame and Parquet sinks.

What rypipe is not

• Not a query engine. No joins, aggregations, window functions, or SQL.
• Not a parser. Each format needs an adapter package.
• Not a data warehouse. It ingests into Arrow; it does not store or serve.
• Not pure Python. Adapters are written in Rust for performance. Python users consume data through adapter APIs; they do not need to write Rust unless creating a new adapter.

Features

• Zero-copy friendly: decoders emit borrowed strings; the engine copies only when necessary.
• GIL-free parsing: heavy work runs outside Python's GIL.
• Parallel by default: chunked parsing with rayon scales to many cores.
• Memory bounded: stream files larger than RAM with iter_record_batches and a configurable budget.
• Pushdown filters: rename, drop, type, and filter rows while parsing.
• Expression API: polars-style col(...) predicates that fuse into the parse loop, composable with &, |, ~.
• Observer hooks: per-row callbacks from the engine, in Rust or Python.
• Transparent decompression: gzip, zstd, and lz4 inputs detected by magic bytes and decompressed automatically.
• Pipeline API: chainable rename/drop/cast/filter stages with automatic fusion.
• Arrow native: produces RecordBatch and exports via the C Data Interface.

Crates

Documentation

• Tutorial: install, first read, pipeline, stages, sinks, streaming
• Writing Adapters: add a new format
• Architecture: how the engine works internally
• Advanced: fusion, execution modes, memory, parallelism
• Python API: full API reference
• Rust API: full API reference

Building

# Rust only
cargo build --workspace --release

# Python extension
maturin develop --release

Testing

# Rust
cargo test --workspace --all-features

# Python
pip install -e ".[dev]"
pytest crates/rypipe-python/tests/

License

MIT

---