Building an Adapter¶

This track teaches you how to write a rypipe adapter, a package that lets rypipe read your custom format.

How this track works¶

Read it in order:

1. Walkthrough, build a complete working adapter (rypipe_log) from scratch, step by step.
2. Python wiring, the full wiring reference: Source, adapter class, registration, stage and sink re-exports, adapter kwargs, engine selection, streaming.
3. Deep dives, one page per component: Splitter, RecordParser, Sink, Schema, scan primitives, and more.

The rest of this page is the map: what you will build, the API contract, and how the engine works.

What you will build¶

A complete adapter package that:

1. Parses a newline-delimited key=value log format.
2. Registers with rypipe so rypipe.read("file.log") works.
3. Supports the full pipeline API (| operator, fusion, streaming).

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

The crxml formula¶

The reference adapter (crxml) defines the standard pattern. Every adapter should follow this structure:

Rust layer¶

Two traits that define your format's parsing logic:

The engine provides TableBuilder as the production ColumnarSink. You rarely implement it yourself.

Python layer¶

Adapter API contract¶

Every adapter must expose these APIs:

Source class¶

from rypipe import Source

class LogSource(Source):
    def _read_arrow(self, plan_overrides=None):
        plan = self._build_plan_kwargs()
        if plan_overrides:
            plan.update(plan_overrides)
        return _rypipe_log.read(str(self._path), **plan)

The Source class gives users the pipeline | operator, caching, and all sinks (.to_arrow(), .to_pandas(), .to_polars(), .to_parquet()).

Stages¶

Re-export these stage classes from rypipe.stages:

• CastTypes, cast column types
• FilterRows, filter rows by predicate
• RenameFields, rename columns
• DropFields, remove columns

See Python Wiring for the re-export pattern.

Sinks¶

Re-export or reimplement these sink functions:

• collect(pipeline), collect to list of dicts
• to_arrow(pipeline), materialize to pyarrow.Table
• to_pandas(pipeline), convert to pandas DataFrame
• to_polars(pipeline), convert to Polars DataFrame
• to_parquet(pipeline, path), write to Parquet
• to_csv(pipeline, path), write to CSV

Registration¶

Register the adapter at import time so rypipe.read("file.ext") works:

def _register():
    try:
        import rypipe
    except Exception:
        return
    rypipe.register_adapter("log", LogAdapter(), extensions=[".log"])

_register()

What NOT to implement¶

Do not implement a read() convenience function. The Source class IS the primary API. Users write:

from my_adapter import MySource

source = MySource("file.myfmt")
table = source.to_arrow()

Not:

from my_adapter import read  # don't do this
table = read("file.myfmt")

User API¶

End users should only import from the adapter package. Here is what a user of your adapter sees:

from rypipe_log import LogSource, CastTypes, FilterRows

source = LogSource("data.log")

# One-liner
table = source.to_arrow()

# Pipeline
result = (
    source
    | CastTypes({"age": int})
    | FilterRows(field="active", op="==", value="true")
).to_arrow()

Users never write from rypipe import CastTypes: they write from rypipe_log import CastTypes. This is the crxml formula.

How the engine works¶

Your adapter provides the parsing logic. The engine handles everything else:

Input bytes (file or mmap)
  │
  ▼
Splitter::next_record_start    (find safe chunk boundaries)
  │
  ▼  [one chunk]
RecordParser::parse_chunk      (per-chunk, feeds ColumnarSink)
  │  calls: begin_row → put_field × N → end_row
  ▼
ColumnarSink (TableBuilder)    (accumulates typed columns)
  │
  ▼
Arrow RecordBatch              (zero-copy export)
  │
  ▼
pyarrow.Table                  (Python API)

rypipe handles:

• Parallel execution: split the file, parse chunks concurrently on multiple threads.
• Bounded-memory streaming: process one chunk at a time, keeping only the current chunk in memory.
• Pushdown plans: rename, drop, filter, type coercion, dictionary encoding, all pushed into the Rust parse loop.
• Zero-copy Arrow export: column buffers move directly into Arrow arrays.
• Schema discovery: find field names from a sample of the file.

Guide contents¶

Performance model¶

The hot path is:

parse_chunk → begin_row → [put_field × N] → end_row → [repeat]

Each put_field call goes through:

1. Scan: find the field's byte extent in the input (your parser does this).
2. Resolve: map raw name to output column name (engine does this).
3. Push: write the value into the column builder (engine does this).
4. Filter: check if the row passes the predicate (engine does this).

The engine optimizes steps 2-4. Your parser's job is to make step 1 fast.

For a 533 MB file on a Ryzen 5800X:

Build and test¶

Every adapter in this guide builds and tests the same way: compile the Rust extension into your environment with uv run --with maturin maturin develop --release, then run the Rust unit tests with cargo test. The walkthrough shows both commands end to end with real output.

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.
• Your parser's job is to make parse_chunk fast.
• Follow the crxml formula for a consistent user experience.