rypipe

Format-agnostic data ingestion framework with Rust core and Python bindings.

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.

Parse row-oriented byte streams into Apache Arrow record batches 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 pure ingestion-to-Arrow engine. It separates format-specific parsing (splitting, row extraction) from format-agnostic execution (typed column builders, filtering, projection, dictionary encoding, parallel scheduling, memory-bounded execution, and Arrow export). Add a new format by implementing two small traits: Splitter and RecordParser.

rypipe itself does not ship parsers for XML, JSON, CSV, HTML, or any other format. Those live in separate adapter packages. Install the engine plus the adapters you need.

Note: crxml was the original idea: a fast Crystal Reports XML parser that needed parallel, bounded-memory, and Arrow-native execution. The engine that made crxml fast (Splitter + RecordParser + TableBuilder + ExecutionPlan) was then separated and abstracted into rypipe so any format could reuse it. crxml now lives as a thin adapter (crxml-core + CrystalXMLSource) on top of rypipe-core. See rypipe docs/crxml-adapter.md for the 3 GB/s evolution.

Why rypipe

• One runtime, many formats. XML, JSON, CSV, HTML, TSV, and any future format share the same parallel scheduler, memory-bounded executor, Arrow export, and pushdown infrastructure. An adapter is two small traits, not a full engine.

• Performance without compromise. Single-thread ~1 GB/s, parallel ~4.9 GB/s unprojected, ~6.8-7.0 GB/s with projection. Zero-copy Arrow export. Predicate first evaluation. Layout prediction via memcmp.

• Correctness by construction. Differential testing against an independent oracle, fuzz targets, property tests, and a tier-ladder profiler that decomposes every nanosecond of the hot path.

• 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. It handles projection, renaming, dropping, casting, filtering, and dictionary encoding. It does not do joins, aggregations, window functions, or SQL.

• Not a one-size-fits-all parser. Each format needs a RecordParser + Splitter adapter from a separate package. The engine provides the runtime; you provide the format knowledge.

• Not a data warehouse. It ingests data into Arrow; it does not store it, index it, or serve queries over it.

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 a configurable budget.
• Typed columns: cast strings to int64, float64, or bool during parse.
• Pushdown filters: rename, drop, type, and filter rows while parsing.
• Pipeline API: chainable rename/drop/cast/filter stages with automatic fusion.
• Dictionary encoding: explicit or automatic low-cardinality encoding.
• Arrow native: produces RecordBatch and exports via the C Data Interface.

Crates

Python quick start

pip install rypipe my-adapter

Wheels are built against CPython's stable ABI (abi3, 3.10+), so one wheel per platform covers every supported interpreter, including versions released after a given rypipe release. Prebuilt wheels ship for manylinux (glibc 2.17+), musllinux, macOS (x86_64 and arm64), and Windows x64; anything else builds from the sdist and needs a Rust toolchain.

Optional DataFrame sinks pull their own dependencies:

pip install "rypipe[pandas]"   # to_pandas / to_dataframe
pip install "rypipe[polars]"   # to_polars
pip install "rypipe[all]"      # both

import rypipe
import my_adapter

table = rypipe.read(
    "data.myfmt",
    fields={"amount": "float64", "qty": "int64"},
    filter={"field": "status", "op": "==", "value": "active"},
)
print(table.num_rows, table.num_columns)

# Bounded-memory streaming.
table = rypipe.read_stream("huge.myfmt", memory="256MiB")

Pipeline API

Adapters that expose a rypipe.Adapter subclass give you a chainable pipeline with automatic fusion of rename, drop, cast, and filter stages into the Rust parse loop. Subclasses only implement read(path, **kwargs)::

from rypipe import RenameFields, DropFields, CastTypes, FilterRows
import my_adapter

source = my_adapter.MySource("data.myfmt")

df = (
    source
    | RenameFields({"old_name": "new_name"})
    | DropFields(["internal_id"])
    | CastTypes({"amount": float, "qty": int})
    | FilterRows(field="status", op="==", value="active")
).to_dataframe()

The same operations work as kwargs on rypipe.read when you only need a table.

Rust quick start

use rypipe_core::{ExecutionPlan, FieldType, Pipeline};
use my_adapter::{MySplitter, MyDecoder}; // separate adapter crate

let batch = Pipeline::new(MySplitter::new(), MyDecoder::new())
    .with_plan(
        ExecutionPlan::new()
            .type_as("amount", FieldType::Float64)
            .type_as("qty", FieldType::Int64)
            .filter_eq("status", "active"),
    )
    .read_path("data.myfmt", false, false)?;

Building

# Rust only
cargo build --workspace --release

# Python extension
maturin develop --release

Testing

# Rust
cargo test --workspace --all-features
cargo clippy --workspace --all-targets --all-features : -D warnings

# Python (against an installed build)
pip install -e ".[dev]"
pytest crates/rypipe-python/tests/

Tests covering optional dependencies (pandas, polars) skip when those packages are absent. Set RYPIPE_REQUIRE_OPTIONAL_DEPS=1 to turn a missing optional dependency into a hard failure instead. CI sets it so that optional coverage cannot silently disappear from a green run.

Benchmark: parallel streaming with frozen schema (parallel Discovery)

crxml (reference adapter) on Ryzen 5800X, 533 MB real / 1 GB synthetic, warm, median-of-7, frozen schema (FrozenSchema crates/rypipe-core/src/schema.rs:21, discovery_ns in get_par_profile()):

* 1 GB auto still 3782/3863 from before parallel Discovery was re-measured on 533 MB only (parallel 19→5.3 ms). Before frozen schema, auto was 4770/4551 (unstable: batch 2 order FieldG vs Text20 last, pq.ParquetWriter raised). Frozen schema fixes order (every batch same, sparse FieldG 30%/Text21 1% as all-null) via ensure_schema crates/rypipe-core/src/engine.rs:79 but adds 5.3 ms (was 15% at 19 ms, now 4% at 5.3 ms), making auto +0.6% vs par128 (4497 vs 4470, within CoV), unblocking auto default. Explicit still +11% and defines the ceiling. Sweep: par peaks at 4 MB, streaming at 2 MB; one divisor cannot serve both, kept split (par 4 MB, streaming 2 MB via budget/(threads×2)). Cap raised 8×threads→16×threads (256) so 533 MB now hits ideal 133 for 4 MB (was capped 128). See crxml docs/performance.md for like-for-like, chunk-per-cell, fixed-chunk isolation (par 1 MB collapses 3553 vs streaming 3812, +7% via chunk_buf reuse).

Run the engine throughput benchmark:

cargo run --release -p rypipe-core --example bench_throughput

Documentation

Full docs and integration guides are in the docs/ directory:

• Overview
• Architecture
• Python API
• Rust API
• Writing a format adapter
• Performance

Why Python, not pure Rust

Data work lives in Python : pip, notebooks, and the PyArrow/pandas/Polars ecosystem. ETL is glue: S3 → parse → rename/drop/cast/filter → validate → write Parquet. That glue is Python; the hot loop that touches every byte at 700 MB/s (2.5 GB/s parallel) must be Rust.

rypipe is Rust where it counts, Python where it ships:

• Zero-copy, GIL-free Rust core parses outside the GIL and hands Arrow RecordBatches to pyarrow via the C Data Interface : no copy, no serialization.
• Python composition : source | RenameFields | FilterRows | CastTypes | .to_dataframe() : lets the same pipeline be explored in a notebook and scaled unchanged in Airflow/Dagster on 100 GB of files.
• Pure-Rust still works when you need it: rypipe-core has no Python dependency and is usable as a crate (Pipeline::new(Splitter, Parser) → read_path_par).

A pure-Rust library would save ~0.1 ms of orchestration and cost the 90% of users who have never run cargo build their workflow. For data teams, a pip install beats a toolchain install every time.

polars didn’t win by being pure Rust : it won with a Rust engine behind pl.DataFrame. rypipe does the same for ingestion. See the full data-driven justification in Why Python?.

License

MIT