rypipe

Format-agnostic columnar ingestion engine with Rust core and Python bindings.

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: the engine model is taken from emiliano-go/crxml, abstracted away from any single format.

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

Crate	Purpose
rypipe-core	Pure Rust engine: Value, ExecutionPlan, TableBuilder, ColumnarSink, RecordParser, Splitter, Pipeline, parallel/bounded drivers, Arrow export
rypipe-python	PyO3 bindings and helper functions for adapter packages; exposes the rypipe package

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

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 rypipe

The engine started out welded to a single format. Abstracting it keeps that speed while making the same execution model available for JSON, CSV, HTML, and other formats through a small adapter interface. Format-specific code now lives in separate packages; this repository contains only the engine and its Python bindings.

License

MIT