Building an Adapter¶

This page shows how to build a rypipe adapter for a newline-delimited key=value log format. By the end, you will have a complete adapter that works with the pipeline API.

The format¶

Your format is newline-delimited key=value pairs:

name=Alice,age=30,active=true
name=Bob,age=25,active=false

Each line is a row. Fields are comma-separated key=value pairs.

Prerequisites¶

• Rust toolchain (1.78+)
• Python 3.10+
• rypipe installed (pip install rypipe)

Step 1: Create the package¶

An adapter is a separate package that depends on rypipe-core. Create the package structure and Cargo.toml:

mkdir rypipe-log && cd rypipe-log
mkdir src

The Cargo.toml defines the Rust crate that will be compiled into a Python extension module. We depend on rypipe-core for the Splitter and RecordParser traits, and pyo3 for Python bindings:

Cargo.toml¶

[package]
name = "rypipe-log"
version = "0.1.0"
edition = "2021"

[lib]
name = "_rypipe_log"
crate-type = ["cdylib"]

[dependencies]
rypipe-core = "2"
pyo3 = { version = "0.29", features = ["extension-module", "abi3-py310"] }
memchr = "2"
simdutf8 = "0.1"

The cdylib crate type produces a shared library that Python can import. The abi3 feature enables stable ABI, so one wheel works across Python versions.

Step 2: Implement the Splitter¶

The Splitter tells the engine where each row starts. The engine calls next_record_start repeatedly to split the file into chunks for parallel parsing. For newline-delimited formats, the next row starts after the next \n:

src/lib.rs (Splitter)¶

use rypipe_core::{Splitter, RecordParser, ColumnarSink, Value, Result};

// The Splitter tells the engine where each row starts.
// For newline-delimited formats, the next row starts after the next '\n'.
#[derive(Clone, Default)]
pub struct LogSplitter;

impl Splitter for LogSplitter {
    // Find the byte position of the next record start after `from`.
    // Return None when we reach the end of the input.
    fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize> {
        memchr::memchr(b'\n', &bytes[from..]).map(|r| from + r + 1)
    }

    // Estimate bytes per row from a sample. The engine uses this to size
    // chunks and memory budgets. Count newlines and divide.
    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
        let n = sample.iter().filter(|&&b| b == b'\n').count().max(1);
        (sample.len() / n).max(1)
    }
}

The Splitter has two methods:

• next_record_start: called repeatedly to find chunk boundaries. The engine splits the file at these positions for parallel parsing. We use memchr for fast newline scanning.
• estimate_bytes_per_row: tells the engine how many rows to expect per chunk, so it can size memory budgets. We count newlines in a sample and divide.

Step 3: Implement the RecordParser¶

The RecordParser extracts field values from each row. This is the hot path: it is called once per chunk, so it must be fast. For each row, we call sink.begin_row(), then sink.put_field() for each field, then sink.end_row():

src/lib.rs (RecordParser)¶

// The RecordParser turns raw bytes into field/value events.
// parse_chunk is called once per chunk: this is the hot path.
#[derive(Clone, Default)]
pub struct LogParser;

impl RecordParser for LogParser {
    // Validate UTF-8 before parsing. Called once per chunk.
    fn validate(&self, bytes: &[u8]) -> Result<()> {
        simdutf8::basic::from_utf8(bytes)
            .map_err(|e| rypipe_core::Error::Utf8(e))?;
        Ok(())
    }

    // Parse a chunk of bytes into field/value events.
    // For each row: begin_row -> put_field x N -> end_row.
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        let text = std::str::from_utf8(bytes)
            .map_err(|e| rypipe_core::Error::Plan(e.to_string()))?;

        for line in text.lines() {
            if line.is_empty() { continue; }

            // Signal the start of a new row
            sink.begin_row();

            // Parse comma-separated key=value pairs
            for part in line.split(',') {
                if let Some((key, value)) = part.split_once('=') {
                    // sink.wants() returns false if the engine doesn't need
                    // this field (projection pushdown). Skip it entirely.
                    if sink.wants(key) {
                        // Borrow the string from the input bytes (zero allocation)
                        sink.put_field(key, Value::Str(std::borrow::Cow::Borrowed(value)));
                    }
                }
            }

            // Signal the end of the row
            sink.end_row();
        }

        Ok(())
    }
}

The RecordParser has two methods:

• validate: called once per chunk to check that the bytes are valid UTF-8. We use simdutf8 for fast validation.
• parse_chunk: the hot path. For each row, call sink.begin_row(), then sink.put_field() for each field, then sink.end_row(). We use Cow::Borrowed to borrow the string from the input bytes without allocation.

Step 4: Expose to Python¶

Add PyO3 bindings to expose your parser to Python. The read_log function is an internal function that LogSource._read_arrow() calls. Users never call it directly, they use LogSource instead:

src/lib.rs (Python bindings)¶

use pyo3::prelude::*;
use rypipe_core::{ExecutionPlan, Pipeline};
use rypipe_python::record_batches_to_pyarrow_table;

// Internal function: called by LogSource._read_arrow().
// Users never call this directly.
#[pyfunction]
fn read_log(path: String) -> PyResult<PyObject> {
    let plan = ExecutionPlan::new();
    let batches = Pipeline::new(LogSplitter, LogParser)
        .with_plan(plan)
        .read_path(&path, false, false)
        .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?;

    Python::with_gil(|py| {
        record_batches_to_pyarrow_table(py, &[batches])
            .map(|obj| obj.into())
    })
}

// Python module definition
#[pymodule]
fn _rypipe_log(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(read_log, m)?)?;
    Ok(())
}

This creates a Python module _rypipe_log with an internal read_log function. The user-facing API is LogSource, which calls read_log internally via _read_arrow().

Step 5: Create the Python wrapper¶

Follow the crxml formula: a Source subclass, a thin adapter, and repacked stages. The Source subclass gives users the pipeline | operator and caching. The thin adapter enables rypipe.read(). The repacked stages make the adapter self-contained.

rypipe_log/__init__.py¶

import importlib

# Side-effect import: registers the adapter with rypipe on import
from . import rypipe_adapter  # noqa: F401

__all__ = [
    "LogSource",
    "LogAdapter",
    "CastTypes",
    "FilterRows",
    "RenameFields",
    "DropFields",
]

_modules = {
    "LogSource": ".source",
    "CastTypes": ".stages",
    "FilterRows": ".stages",
    "RenameFields": ".stages",
    "DropFields": ".stages",
}


def __getattr__(name):
    if name in _modules:
        mod = importlib.import_module(_modules[name], __package__)
        return getattr(mod, name)
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")


def __dir__():
    return __all__

The __init__.py uses lazy loading: modules are only imported when accessed. This avoids loading the Rust extension until it is actually needed.

rypipe_log/source.py¶

The Source subclass is the pipeline-capable entry point. It implements _read_arrow() and forwards plan kwargs from fused stages:

from typing import Any

import _rypipe_log
from rypipe import Source


class LogSource(Source):
    """Pipeline-capable source for newline-delimited key=value logs."""

    def _read_arrow(self, plan_overrides: dict[str, Any] | None = None) -> Any:
        # Start with construction-time kwargs (field_mapping, drop_fields, etc.)
        plan = self._build_plan_kwargs()
        # Fused pipeline stages override construction-time kwargs
        if plan_overrides:
            plan.update(plan_overrides)
        # Pass the merged plan to the Rust reader
        return _rypipe_log.read_log(str(self._path), **plan)

rypipe_log/rypipe_adapter.py¶

The adapter inherits from rypipe.Adapter, which handles plan forwarding automatically. Subclasses only implement read():

from typing import Any

from rypipe import Adapter


class LogAdapter(Adapter):
    """rypipe-compatible adapter for newline-delimited key=value logs."""

    def read(self, path: str, **kwargs: Any) -> Any:
        """Parse ``path`` and return a ``pyarrow.Table``."""
        return _rypipe_log.read_log(path, **kwargs)


def _register() -> None:
    try:
        import rypipe
    except Exception:  # pragma: no cover, rypipe is optional
        return
    rypipe.register_adapter("log", LogAdapter(), extensions=[".log"])


_register()

For the full stage implementations (CastTypes, FilterRows, etc.), see Stages.

Step 6: Build and test¶

Build the Rust extension with maturin, then test your adapter:

Build¶

pip install maturin
maturin develop --release

Test it¶

import rypipe
import rypipe_log  # registers the adapter

# Create a test file
with open("test.log", "w") as f:
    f.write("name=Alice,age=30,active=true\n")
    f.write("name=Bob,age=25,active=false\n")

# Pattern 1: one-liner via rypipe (extension auto-detected)
table = rypipe.read("test.log")
print(table)
# pyarrow.Table<name: string, age: string, active: string>

# Pattern 2: pipeline via LogSource + repacked stages
from rypipe_log import LogSource, CastTypes, FilterRows

src = LogSource("test.log")
result = (
    src
    | CastTypes({"age": int})
    | FilterRows(field="active", op="==", value="true")
).to_arrow()
print(result)
# pyarrow.Table<name: string, age: int64, active: string>
# name: ["Alice"]
# age: [30]
# active: ["true"]

Pattern 2 (using the Source directly) is the recommended approach. It gives you the pipeline | operator, caching, and streaming. Pattern 1 (rypipe.read()) is a convenience for one-liner reads but does not support pipelines.

What just happened¶

1. Splitter found newline boundaries in the file.
2. RecordParser parsed each chunk, calling sink.put_field for each field in each row.
3. Engine accumulated values into Arrow columns.
4. Export produced a pyarrow.Table with zero-copy.

Next steps¶

• Pipeline, how the | operator and plan forwarding work
• Stages, implement CastTypes, FilterRows, etc.
• Sinks, implement collect, to_pandas, to_csv
• Rust Creation, deep dive into Splitter, RecordParser, and ColumnarSink
• Schema, declare columns for maximum performance

Recap¶

• An adapter is a Rust crate (Splitter + RecordParser) and a Python package (Source + stages + sinks).
• The engine handles parallel execution, memory management, and Arrow export.
• Users import everything from the adapter package, never from rypipe directly.
• rypipe.read("file.log") works via the registered adapter.
• LogSource("file.log") | CastTypes(...) | FilterRows(...) works via the Source pipeline.