# rypipe Documentation
> Full documentation for rypipe: format-agnostic columnar ingestion engine with Rust core and Python bindings
> Source: https://rypipe.emiliano-go.com
> Pages: 6

========================================================================
PAGE: https://rypipe.emiliano-go.com/architecture/
========================================================================

# Architecture

`rypipe` is built around one idea: **separate the parts of parsing that depend
on a file format from the parts that don't.**

The format-specific side answers questions like:

- Where does one row end and the next begin?
- How do I extract field names and values from a row?
- What is the encoding / entity-escaping rule?

The format-agnostic side answers questions like:

- How do I store rows as typed columns?
- How do I rename, drop, cast, filter, and reorder columns?
- How do I export to Arrow?
- How do I parallelize parsing while staying inside a memory budget?

## High-level flow

```
input bytes
    │
    ▼
┌─────────────────┐
│   Splitter      │  ← format-specific: finds safe chunk boundaries
└────────┬────────┘
         │ Vec<Range<usize>>
         ▼
┌─────────────────┐
│ RecordParser    │  ← format-specific: turns bytes into field events
│   begin_row()   │
│   put_field()   │
│   end_row()     │
└────────┬────────┘
         │ Value events
         ▼
┌─────────────────┐
│  TableBuilder   │  ← format-agnostic: columnar storage + plan
│ (ColumnarSink)  │
└────────┬────────┘
         │ RecordBatch
         ▼
┌─────────────────┐
│  Arrow export   │  ← format-agnostic: C Data Interface / compute kernels
└─────────────────┘
```

## Crate overview

### `rypipe-core`

The pure-Rust engine. It has no `pyo3` or `quick-xml` dependency and no
format-specific logic.

| Module | Responsibility |
|--------|----------------|
| `value` | `Value<'a>` enum: `Str(&str)`, `Int64`, `Float64`, `Bool`, `Null`. |
| `plan` | `ExecutionPlan`, `FieldType`, `FilterPredicate`, `CompareOp`. |
| `columnar` | `StrColumn`, `ColumnBuilder`, dictionary encoding, auto-dict heuristic. |
| `engine` | `TableBuilder`, the main `ColumnarSink` implementation. |
| `merge` | `TableBuilder::extend`, `engines_to_record_batches`. |
| `arrow_export` | Build Arrow arrays, apply `Compare` filters via `arrow::compute`. |
| `decoder` | `Splitter`, `RecordParser`, `ColumnarSink` traits. |
| `pipeline` | `Pipeline` helper that wires a `Splitter` + `RecordParser` to the engine. |
| `parallel` | `ParallelExecutor` over a `Splitter` + `RecordParser`. |
| `bounded` | `BoundedExecutor` + `MemoryBudget` for streaming large files. |
| `input` | `InputBuffer` abstraction: mmap or owned `Vec<u8>`. |
| `error` | Unified `Error` / `Result` type. |

### Adapter packages (not shipped with rypipe)

Format-specific adapters are separate packages. An adapter provides:

- A `RecordParser` implementation that turns bytes into field events.
- A `Splitter` implementation that finds safe chunk boundaries.

For example, an XML adapter lives in its own package and provides an
`XmlSplitter` and an `XmlDecoder` that implement these traits.

### `rypipe-python`

PyO3 bindings and Python package glue for the engine. It provides:

- The `rypipe` Python package:
  - `Adapter` and `Source` base classes for adapter packages.
  - Fusable pipeline stages: `RenameFields`, `DropFields`, `CastTypes`, `FilterRows`.
  - Sinks: `collect`, `to_arrow`, `to_dataframe`, `to_pandas`, `to_polars`, `to_parquet`, `to_csv`.
  - An adapter registry (`register_adapter`, `read`, `read_par`, `read_stream`)
    and the shared exception types.
- The `_rypipe` native extension: typed exceptions (`ParseError`, `XmlError`,
  `PlanError`, `MergeError`) and Rust helper functions
  (`execution_plan_from_kwargs`, `record_batches_to_pyarrow_table`) that adapter
  crates use to build their own Python APIs.

## The decoder API

The boundary between format-specific and format-agnostic code is three traits.

### `Splitter`

```rust
pub trait Splitter: Send + Sync {
    fn find_split_points(&self, bytes: &[u8], max_chunks: usize) -> Vec<usize>;
    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize;
}
```

`find_split_points` returns sorted byte offsets. Adjacent offsets define a
chunk range. The first offset should be `0` and the last should be
`bytes.len()`. A good splitter guarantees that each chunk starts at a valid
row boundary so chunks can be parsed independently.

`estimate_bytes_per_row` is used by the bounded executor to size batches.

### `RecordParser`

```rust
pub trait RecordParser: Send + Sync {
    fn validate(&self, bytes: &[u8]) -> Result<()>;
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()>;
}
```

`validate` checks well-formedness once per chunk (e.g. UTF-8 validity). The
main parse loop then calls `begin_row`, `put_field`, and `end_row` on the sink.

`RecordParser` never sees the `ExecutionPlan`. It only resolves native field
identities to plain names. For XML that means emitting the attribute/child name;
for CSV it means mapping a header index to a column name internally.

### `ColumnarSink`

```rust
pub trait ColumnarSink {
    fn begin_row(&mut self);
    fn put_field(&mut self, name: &str, value: Value<'_>);
    fn end_row(&mut self);
    fn wants(&self, _name: &str) -> bool { true }
    fn finish(&mut self) -> Result<RecordBatch>;
}
```

`TableBuilder` is the canonical implementation. It:

1. Resolves the field name through `ExecutionPlan::resolve_field` (rename-then-drop).
2. Ensures a `ColumnBuilder` exists for the resolved name.
3. Applies **last-write-wins** within the current row.
4. On `end_row`, null-fills missing columns and evaluates per-row filters.
5. On `finish`, sorts columns by `schema_order`, runs auto-dict upgrade, and
   builds Arrow arrays.

`wants` lets parsers skip fields that will be dropped, avoiding wasted
extraction work.

## Execution plan

`ExecutionPlan` is the format-agnostic pushdown target.

```rust
pub struct ExecutionPlan {
    pub field_map: HashMap<String, String>,       // rename
    pub drop_fields: HashSet<String>,             // drop
    pub field_types: HashMap<String, FieldType>,  // cast
    pub dictionary_columns: HashSet<String>,      // explicit dict encoding
    pub filter: Option<FilterPredicate>,          // per-row or post-reduce
    pub schema_order: Vec<String>,                // output column order
    pub auto_dict: bool,                          // auto-dict upgrade
}
```

Field resolution order:

1. `field_map` renames the raw field.
2. `drop_fields` checks the resolved name.
3. `field_types` / `dictionary_columns` chooses the storage type.
4. `filter` rejects rows during `end_row` (for `Equal`/`NotEqual`) or after
   assembly (for `Compare`).

`FilterPredicate::Compare` is evaluated after the `RecordBatch` is built using
`arrow::compute` comparison kernels and `filter_record_batch`. This removes the
previous dependency on calling `pyarrow.compute` from inside Rust.

## Columnar storage

### `StrColumn`

Strings are stored in one contiguous byte arena plus `i32` offsets and a
validity bitmap. This is exactly the Arrow string layout, so export is a block
copy of two buffers.

- `push` appends bytes and an offset; no per-cell `String` allocation.
- `pop` truncates the arena; enables row-level filtering without compaction.
- `append` merges another column by base-shifting offsets.

### `ColumnBuilder`

An enum over storage types:

```rust
pub(crate) enum ColumnBuilder {
    String(StrColumn),
    Int64(Vec<Option<i64>>),
    Float64(Vec<Option<f64>>),
    Boolean(Vec<Option<bool>>),
    Dictionary { codes: Vec<Option<i32>>, dict: Vec<String>, index: HashMap<String, i32> },
}
```

Typed builders parse from `Value::Str` or accept native typed `Value` variants
directly. Unparseable strings become `None` (null).

Dictionary encoding uses a `value → i32` side index. When two dictionary
columns are merged, the right-hand dictionary is remapped into the left-hand
one and codes are translated in one pass.

## Parallel execution

`ParallelExecutor::parse` does the following:

1. Calls `Splitter::find_split_points`.
2. Converts points to non-empty `Range<usize>` chunks.
3. Uses `rayon::par_iter` to parse each chunk independently into a
   `TableBuilder`.
4. **Fast path**: if `auto_dict` is false and there is no `Compare` filter,
   each builder is exported as its own `RecordBatch` in parallel via
   `engines_to_record_batches`. No serial merge happens.
5. **Merge path**: if `auto_dict` or a `Compare` filter is present, chunk
   builders are merged sequentially, the filter is applied, and a single
   `RecordBatch` is returned.

Panic catching per chunk prevents one malformed chunk from killing the whole
parallel parse.

## Bounded execution

`BoundedExecutor::run` keeps peak memory near a configured budget:

1. Opens the file via `InputBuffer`.
2. Estimates `bytes_per_row` from the splitter.
3. Computes `rows_per_batch` from the budget.
4. Splits the file into at most 64 batches.
5. Parses each batch into a `TableBuilder`, exports it to a `RecordBatch`, and
   resets the builder.
6. Returns a `Vec<RecordBatch>`; the caller concatenates.

Because the input buffer is dropped before the parse phase begins for bounded
mode, mmap-backed pages are released before downstream work starts.

## Memory model

- `InputBuffer::Mmap` maps the file and applies `MADV_WILLNEED` (prefault) or
  `MADV_SEQUENTIAL` (RSS-sensitive) advice on Unix. The mapping is dropped
  before Arrow export, so no borrowed bytes outlive it.
- `InputBuffer::Owned` simply reads the file into a `Vec<u8>`.
- `StrColumn` owns its bytes; Arrow arrays are built from owned buffers.
- Numeric columns use dense `Vec<Option<T>>`.

## Error handling

`rypipe-core` uses one `Error` enum:

```rust
pub enum Error {
    Utf8(...),
    Plan(String),
    Merge(String),
    Io(...),
    Arrow(...),
}
```

Adapters can map their own errors into this type. `rypipe-python` maps these
to `XmlError`, `PlanError`, and `MergeError` Python exceptions.

## Why this shape?

The original crxml engine was fast but tightly coupled to one XML dialect.
Extracting it into rypipe keeps those performance characteristics (arena
string storage, SIMD UTF-8 validation, zero-copy event parsing, GIL release,
parallel chunking, memory bounding) while making them available to other
formats. A new adapter only needs to answer "where are the rows?" and "what are
the fields?"; the engine handles the rest.

## See also

- [Rust API](./rust-api.md): `Pipeline`, `ExecutionPlan`, and custom adapters.
- [Writing a format adapter](./writing-adapters.md): step-by-step adapter guide.
- [Python API](./python-api.md): `rypipe.read` and the adapter registry.

========================================================================
PAGE: https://rypipe.emiliano-go.com/
========================================================================

# rypipe documentation

`rypipe` is a format-agnostic columnar engine that turns byte streams into
Apache Arrow record batches. 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.

`rypipe` itself does **not** ship parsers for any format. Adapters live in
separate packages. Install the engine plus the adapters you need.

## What rypipe is

- A Rust workspace with two crates:
  - [`rypipe-core`](./architecture.md#rypipe-core): the generic engine.
  - [`rypipe-python`](./architecture.md#rypipe-python): PyO3 bindings and helper
    functions for adapter packages.
- Zero-copy friendly: decoders emit borrowed strings; the engine copies only
  when necessary.
- GIL-free parsing: all heavy work runs outside Python's GIL.
- Memory-bounded and parallel by design.

## What rypipe is not

- Not a full query engine. It handles projection, renaming, dropping, casting,
  filtering, and dictionary encoding, not joins, aggregations, or SQL.
- Not a one-size-fits-all parser. Each format needs a `RecordParser` + `Splitter`
  adapter from a separate package.

## Quick start

### From Python

```bash
pip install rypipe my-adapter
```

```python
import rypipe
import my_adapter

# Format is inferred from the extension; mode defaults to parallel.
table = rypipe.read(
    "data.myfmt",
    fields={"amount": "float64"},
    filter={"field": "status", "op": "==", "value": "active"},
)
print(table.num_rows, table.num_columns)
```

### 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)``::

```python
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()
```

### From Rust

```rust
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)
            .filter_eq("status", "active"),
    )
    .read_path("data.myfmt", false, false)?;
```

## Guides

- [Architecture](./architecture.md): how the pieces fit together.
- [Python API](./python-api.md): the `rypipe` package and `_rypipe` helpers.
- [Rust API](./rust-api.md): using `rypipe-core` and writing custom adapters.
- [Writing a format adapter](./writing-adapters.md): adding CSV, JSON, etc.
- [Performance](./performance.md): benchmarks and tuning knobs.

## Repository layout

```
rypipe/
├── Cargo.toml                 # workspace
├── pyproject.toml             # maturin / Python package
├── README.md
├── LICENSE
├── crates/
│   ├── rypipe-core/           # generic engine
│   └── rypipe-python/         # PyO3 bindings and helper functions
└── docs/
    └── (this directory)
```

========================================================================
PAGE: https://rypipe.emiliano-go.com/performance/
========================================================================

# Performance

`rypipe` is designed to keep parsing fast: arena string storage, SIMD UTF-8
validation, zero-copy event parsing, GIL-free Rust work, parallel chunking, and
a bounded-memory streaming path. This page explains how to measure and tune the
engine.

## Measured throughput

The numbers below come from the built-in `bench_throughput` example. It uses a
tiny inline TSV-like adapter so the result measures the engine, not an external
parser.

Run it yourself:

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

Hardware: Linux workstation, AMD Ryzen 9 5900X, DDR4-3200, release build.

| Path | Rows | Time | Rows/s | MB/s |
|------|------|------|--------|------|
| `Pipeline::read_path` | 5,000,000 | 1.34 s | 3.75 M | 224 |
| `Pipeline::read_path_par` (4 chunks) | 5,000,000 | 1.47 s | 3.41 M | 204 |
| `Pipeline::read_path_par` (8 chunks) | 5,000,000 | 1.45 s | 3.44 M | 206 |
| `Pipeline::read_path_stream` (64 MiB) | 5,000,000 | 1.99 s | 2.51 M | 150 |

Memory stayed around **330 MB RSS** for the single-thread and parallel paths.
The bounded streaming path kept intermediate batches near the 64 MiB budget.

## Tuning knobs

### Number of chunks (`num_chunks`)

`ParallelExecutor::parse` accepts `num_chunks`. More chunks improve load
balancing but add scheduling overhead. For most workloads, 2-4 times the number
of logical cores is a good starting point. Very fast parsers (like the simple
TSV adapter above) can become memory-bandwidth bound, so adding chunks beyond a
certain point stops helping.

### Memory budget (`memory`)

`BoundedExecutor` keeps intermediate builder storage under the given byte
budget. It does not include Arrow export or downstream pandas conversion; set
the budget lower than total RAM. A value like 500 MB is reasonable for
workstations.

### `use_mmap` and `prefault`

| Combination | Best for |
|-------------|----------|
| `use_mmap=True, prefault=True` | Speed when the file fits in RAM. |
| `use_mmap=True, prefault=False` | Large files where RSS matters. |
| `use_mmap=False` | Portability; reads into a `Vec<u8>`. |

`prefault=True` uses `MADV_WILLNEED` to fault the whole file up front.
`prefault=False` uses `MADV_SEQUENTIAL` so the kernel can drop pages behind the
reader.

### `auto_dict`

When `auto_dict=True`, string columns with low cardinality are upgraded to
dictionary encoding. This reduces memory and can speed up downstream operations,
but it forces a serial merge of all chunk builders in parallel mode, which
raises peak RSS and removes the fast per-chunk export path.

Use `auto_dict=True` when:

- Columns have many repeated values.
- You need smaller Arrow files or faster group-by/filter operations.

Use `auto_dict=False` when:

- Throughput is the top priority.
- Columns are high cardinality or already numeric.

### `field_types`

Casting strings to numbers during parse avoids storing intermediate strings and
lets numeric `Compare` filters use native kernels. If you know a column is
numeric, declare it.

### `schema`

Providing `schema_order` avoids the small cost of sorting columns at finish time
and makes output column order deterministic across chunks.

## Fast path vs merge path

`ParallelExecutor` has two internal paths:

- **Fast path**: when `auto_dict` is false and there is no `Compare` filter,
  each chunk is exported as its own `RecordBatch` in parallel. No serial merge.
- **Merge path**: when `auto_dict` or a `Compare` filter is enabled, chunk
  builders are merged sequentially before export. Peak RSS is higher.

If you need both a `Compare` filter and maximum throughput, consider filtering
after export in Python/Arrow instead.

## GIL behavior

All parse paths release the GIL during the heavy Rust work. The Arrow C Data
Interface export re-acquires the GIL briefly. For `read_path_par`, the entire
parallel parse runs outside the GIL.

## Profiling

Build with the `profiling` profile for symbols:

```bash
cargo build --profile profiling -p rypipe-core
```

Then use `perf`, `cargo flamegraph`, or `samply` to profile.

## Future work

- A generic streaming `RecordParser` could support chunked async input.
- Dictionary encoding could be made incremental across chunks to recover the
  fast path for `auto_dict=True`.

## See also

- [Architecture](./architecture.md): engine design and fast/merge paths.
- [Rust API](./rust-api.md): tuning `num_chunks` and `memory` from Rust.
- [Python API](./python-api.md): the same knobs from Python.

========================================================================
PAGE: https://rypipe.emiliano-go.com/python-api/
========================================================================

# Python API

`rypipe-python` builds a mixed Rust/Python package. The public API lives in the
`rypipe` package; `_rypipe` is the low-level Rust extension that adapter
packages build on.

`rypipe` itself does **not** ship any format parsers. Install a separate adapter
package and import it; the adapter registers itself with `rypipe` so the
high-level `read` API works. Adapters can also expose a `Source` subclass and
get the pipeline/stage/sink API for free.

## Building the Python module

```bash
export PYO3_PYTHON=/path/to/python3.12
maturin develop --release
```

`maturin` builds `crates/rypipe-python/Cargo.toml` and installs both the `rypipe`
Python package and the `_rypipe` Rust extension.

## Public API (`import rypipe`)

### `rypipe.Source`

Abstract base class for row-oriented file sources. Adapter packages subclass it
and implement `_read_arrow`. Once they do, users get pipelines, stages, and
sinks with no extra work.

```python
from rypipe import Source

class MySource(Source):
    def _read_arrow(self, plan_overrides=None):
        # Build plan from construction kwargs + overrides, call the parser,
        # return a pyarrow.Table.
        ...
```

### `rypipe.Adapter`

Even simpler: subclass `Adapter` and implement only ``read(path, **kwargs)``.
Plan kwargs are merged and passed through automatically.

```python
from rypipe import Adapter

class CsvAdapter(Adapter):
    def read(self, path, **kwargs):
        return _rypipe_csv.read_csv(path, **kwargs)

source = CsvAdapter("data.csv")
```

A `Source` exposes:

- Row iteration: `for row in source`
- Table export: `source.to_arrow()`, `source.to_pandas()`, `source.to_polars()`,
  `source.to_parquet(path)`
- Pipeline operator: `source | RenameFields(...)`
- Caching: `source.clear_cache()`

### Pipeline stages

`rypipe` ships the same fusable stages crxml used. Stages that rename, drop,
cast, or filter constants are pushed into the Rust parse loop when the source
supports plan kwargs.

```python
from rypipe import RenameFields, DropFields, CastTypes, FilterRows

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

`CastTypes` accepts Python callables (`int`, `float`, `bool`, `str`). When the
callable maps to a Rust type (`int64`, `float64`, `bool`), it is fused into the
Rust parse loop.

`FilterRows` supports both constant filters and column-to-column comparisons:

```python
FilterRows(field="status", op="==", value="active")
FilterRows(field_a="amount", op=">", field_b="threshold")
```

Supported ops: `==`, `!=`, `>`, `<`, `>=`, `<=`.

### Pipeline sinks

```python
from rypipe import collect, to_arrow, to_dataframe, to_csv, to_parquet

rows = collect(pipeline)
table = to_arrow(pipeline)
df = to_dataframe(pipeline)
to_csv(pipeline, "out.csv")
to_parquet(pipeline, "out.parquet")
```

Sinks try the fused Arrow path first and fall back to dict iteration when the
pipeline ends with a generic stage.

### `rypipe.read`

Single entry point for all registered adapters.

```python
import rypipe
import my_adapter  # registers the "myfmt" adapter

table = rypipe.read("data.myfmt")

# Same call with all common options:
table = rypipe.read(
    "data.myfmt",
    format="myfmt",          # inferred from extension when omitted
    fields={"amount": "float64", "qty": "int64"},
    dictionary=["status"],
    filter={"field": "status", "op": "==", "value": "active"},
    schema=["id", "status", "amount"],
    auto_dict=False,
    use_mmap=False,
    prefault=False,
)
```

Returns a `pyarrow.Table`.

You can also pass an adapter object directly:

```python
table = rypipe.read("data.myfmt", adapter=my_adapter, row_tag="Row")
```

### `rypipe.read_par`

Convenience wrapper that passes `chunks` to the adapter.

```python
table = rypipe.read_par("data.myfmt", chunks=8, fields={"amount": "float64"})
```

### `rypipe.read_stream`

Convenience wrapper that passes a memory budget to the adapter. `memory`
accepts an int (bytes) or a human-readable string such as `"128MiB"`.

```python
table = rypipe.read_stream("huge.myfmt", memory="500MiB", row_tag="Row")
```

### Format auto-detection

`rypipe.read` infers the adapter from the file extension when `format` is not
provided, but only for extensions registered by an installed adapter package.
If no adapter is registered, pass `format=` explicitly or install the adapter.

### Exceptions

| Exception | Meaning |
|-----------|---------|
| `rypipe.ParseError` | Malformed input or parse failure (including invalid UTF-8). |
| `rypipe.XmlError` | Backward-compatible alias of `ParseError`. |
| `rypipe.PlanError` | Invalid pushdown plan (unknown field type, bad filter op). |
| `rypipe.MergeError` | Chunk-merge conflict (e.g. type mismatch across chunks). |
| `rypipe.RypipeError` | Invalid API usage (bad memory string, unknown extension). |

## Low-level API (`import _rypipe`)

`_rypipe` is the Rust extension that adapter packages build on. It exposes the
shared exceptions and Rust helper functions; adapter packages implement the
actual `read` functions and call these helpers from their own PyO3 code.

### Exceptions

- `_rypipe.ParseError`
- `_rypipe.XmlError`
- `_rypipe.PlanError`
- `_rypipe.MergeError`

Adapter code raises these so users can catch them through `rypipe` as well.

### Rust helpers (used from adapter crates)

Adapter crates written in Rust use `rypipe_python` directly:

```rust
use rypipe_python::{execution_plan_from_kwargs, record_batches_to_pyarrow_table};
```

`execution_plan_from_kwargs` converts Python kwargs into a
`rypipe_core::ExecutionPlan`. `record_batches_to_pyarrow_table` turns a slice of
Arrow `RecordBatch`es into a single `pyarrow.Table`.

## Plan kwargs

All public `read` functions and `Source` constructors accept the same pushdown
kwargs, which are passed through to the adapter.

| Kwarg | Type | Effect |
|-------|------|--------|
| `rename` / `field_mapping` | `dict[str, str]` | Rename raw fields. |
| `drop` / `drop_fields` | `list[str]` | Drop fields by resolved name. |
| `fields` / `field_types` | `dict[str, str]` | Cast columns to `"int64"`, `"float64"`, `"bool"`, `"dictionary"`, or `"string"`. |
| `dictionary` / `dictionary_columns` | `list[str]` | Explicit dictionary encoding. |
| `filter` | `dict` | Per-row or post-reduce filter (see below). |
| `schema` | `list[str]` | Output column order. |
| `auto_dict` | `bool` | Upgrade low-cardinality string columns to dictionary. |
| `use_mmap` | `bool` | Memory-map the input file. |
| `prefault` | `bool` | `MADV_WILLNEED` when mmap is enabled. |

## Filters

Constant equality/inequality (evaluated per-row during parse):

```python
filter={"field": "status", "op": "==", "value": "active"}
filter={"field": "status", "op": "!=", "value": "archived"}
```

Column-to-column comparison (evaluated after the table is assembled):

```python
filter={"field_a": "amount", "op": ">", "field_b": "threshold"}
```

Supported comparison ops: `>`, `<`, `>=`, `<=`, `==`, `!=`.

## See also

- [Rust API](./rust-api.md): the Rust engine and `Pipeline` API.
- [Writing a format adapter](./writing-adapters.md): adding formats as separate packages.
- [Architecture](./architecture.md): design overview.

========================================================================
PAGE: https://rypipe.emiliano-go.com/rust-api/
========================================================================

# Rust API

`rypipe-core` is a pure-Rust crate. This guide shows how to use it directly and
how its pieces compose. Format-specific adapters (for XML, CSV, JSON, etc.) are
separate crates that implement `Splitter` and `RecordParser`.

## Dependencies

```toml
[dependencies]
rypipe-core = { path = "../rypipe/crates/rypipe-core" }
# adapter crate of your choice, e.g.:
# my-csv-adapter = "0.1"
```

## Recommended entry point: `Pipeline`

`Pipeline` wires a `Splitter` and `RecordParser` together and removes the
boilerplate of opening files and choosing an execution mode.

```rust
use rypipe_core::{ExecutionPlan, FieldType, Pipeline};
// Import the adapter for your format from its own crate:
use my_csv_adapter::{CsvSplitter, CsvDecoder};

fn main() -> rypipe_core::Result<()> {
    let pipeline = Pipeline::new(CsvSplitter::new(), CsvDecoder::new())
        .with_plan(
            ExecutionPlan::new()
                .rename("old_name", "new_name")
                .drop("junk")
                .type_as("amount", FieldType::Float64)
                .dictionary("status")
                .filter_eq("status", "active")
                .schema_order(["id", "status", "amount"])
                .with_auto_dict(true),
        );

    // Single-file parse.
    let batch = pipeline.read_path("data.csv", false, false)?;
    println!("rows={} cols={}", batch.num_rows(), batch.num_columns());

    // Parallel parse.
    let batches = pipeline.read_path_par("data.csv", 8, false, false)?;

    // Bounded-memory streaming.
    let batches = pipeline.read_path_stream(
        "huge.csv",
        rypipe_core::MemoryBudget::new(500_000_000),
        false,
    )?;

    Ok(())
}
```

## A minimal example (low level)

If you prefer to control every step, use `TableBuilder` directly:

```rust
use rypipe_core::{InputBuffer, TableBuilder, ExecutionPlan};
use my_csv_adapter::{CsvSplitter, CsvDecoder}; // separate adapter crate

fn main() -> rypipe_core::Result<()> {
    let input = InputBuffer::open("data.csv".as_ref(), false, false)?;
    let mut builder = TableBuilder::with_plan(1024, ExecutionPlan::new());

    let decoder = CsvDecoder::new();
    decoder.validate(input.as_slice())?;
    decoder.parse_chunk(input.as_slice(), &mut builder)?;

    let batch = builder.finish()?;
    println!("rows={} cols={}", batch.num_rows(), batch.num_columns());
    Ok(())
}
```

## `ExecutionPlan`

The builder API is the recommended way to construct a plan:

```rust
use rypipe_core::{CompareOp, ExecutionPlan, FieldType};

let plan = ExecutionPlan::new()
    .rename("old_name", "new_name")
    .drop("junk")
    .type_as("amount", FieldType::Float64)
    .dictionary("status")
    .filter_eq("status", "active")
    .filter_compare("amount", CompareOp::Gt, "threshold")
    .schema_order(["id", "status", "amount"])
    .with_auto_dict(true);
```

You can still mutate the fields directly when you need to:

```rust
let mut plan = ExecutionPlan::new();
plan.field_map.insert("old_name".into(), "new_name".into());
plan.drop_fields.insert("junk".into());
plan.field_types.insert("amount".into(), FieldType::Float64);
plan.dictionary_columns.insert("status".into());
plan.filter = Some(rypipe_core::FilterPredicate::Equal {
    field: "status".into(),
    value: "active".into(),
});
plan.schema_order = vec!["id".into(), "status".into(), "amount".into()];
plan.auto_dict = true;
```

## `Value`

Decoders emit `Value<'a>`:

```rust
use rypipe_core::Value;

sink.put_field("amount", Value::Float64(123.45));
sink.put_field("name", Value::Str("Alice"));
sink.put_field("flag", Value::Bool(true));
sink.put_field("missing", Value::Null);
```

For stringly formats everything is a string, but JSON or CSV adapters can emit
native typed values and skip string parsing.

## Parallel parse (low level)

```rust
use rypipe_core::{parallel::ParallelExecutor, ExecutionPlan};
use my_csv_adapter::{CsvSplitter, CsvDecoder}; // separate adapter crate

let bytes = std::fs::read("data.csv")?;
let splitter = CsvSplitter::new();
let decoder = CsvDecoder::new();
let plan = ExecutionPlan::new();

let batches = ParallelExecutor::parse(&bytes, &splitter, decoder, plan, 8)?;
```

`ParallelExecutor::parse` returns a `Vec<RecordBatch>`. The fast path emits one
batch per chunk; the merge path returns a single merged batch when `auto_dict`
or a `Compare` filter is enabled.

## Bounded parse (low level)

```rust
use rypipe_core::{
    bounded::{BoundedExecutor, MemoryBudget},
    ExecutionPlan,
};
use my_csv_adapter::{CsvSplitter, CsvDecoder}; // separate adapter crate
use std::path::Path;

let splitter = CsvSplitter::new();
let decoder = CsvDecoder::new();
let batches = BoundedExecutor::new(MemoryBudget::new(500_000_000))
    .run(Path::new("huge.csv"), &splitter, decoder, ExecutionPlan::new(), false)?;
```

## Apply a post-reduce Compare filter

```rust
use rypipe_core::{apply_compare_filter, FilterPredicate, CompareOp};

let predicate = FilterPredicate::Compare {
    field_a: "amount".into(),
    op: CompareOp::Gt,
    field_b: "threshold".into(),
};
let filtered = apply_compare_filter(batch, &predicate)?;
```

## Export helpers

`rypipe-python` provides Rust helper functions for adapter crates:

```rust
use rypipe_python::{execution_plan_from_kwargs, record_batches_to_pyarrow_table};

// Inside a PyO3 function:
let plan = execution_plan_from_kwargs(...)?;
let table = record_batches_to_pyarrow_table(py, &batches)?;
```

For a pure-Rust program you do not need this; `arrow::record_batch::RecordBatch`
is already sufficient.

## Writing a custom sink

You can implement `ColumnarSink` yourself for specialized behavior. Most users
should use `TableBuilder`.

```rust
use rypipe_core::{ColumnarSink, Value, Result};
use arrow::record_batch::RecordBatch;

struct RowCounter { rows: usize }

impl ColumnarSink for RowCounter {
    fn begin_row(&mut self) {}
    fn put_field(&mut self, _name: &str, _value: Value<'_>) {}
    fn end_row(&mut self) { self.rows += 1; }
    fn finish(&mut self) -> Result<RecordBatch> {
        Ok(RecordBatch::new_empty(std::sync::Arc::new(
            arrow::datatypes::Schema::empty(),
        )))
    }
}
```

## See also

- [Writing a format adapter](./writing-adapters.md): implement `Splitter` and `RecordParser` in a separate package.
- [Architecture](./architecture.md): how the pieces fit together.
- [Python API](./python-api.md): the Python bindings over the same engine.

========================================================================
PAGE: https://rypipe.emiliano-go.com/writing-adapters/
========================================================================

# Writing a format adapter

A `rypipe` adapter is just two small types: a `Splitter` and a `RecordParser`.
Once you have those, the `Pipeline` API wires them into one-file, parallel, and
bounded-memory execution with one line of code.

Adapters are **separate packages**, not part of `rypipe`. This keeps `rypipe`
pure: it is only the ingestion-to-Arrow engine. Your adapter crate depends on
`rypipe-core` and, if you want Python bindings, on `rypipe-python` for the
plan/export helpers.

## Adapter crate layout

```
rypipe-csv/
├── Cargo.toml
├── pyproject.toml            # optional, for a Python package
└── src/
    ├── lib.rs
    ├── splitter.rs
    └── decoder.rs
```

```toml
[dependencies]
rypipe-core = "0.1"

# Only if you build Python bindings for the adapter:
rypipe-python = "0.1"
pyo3 = { version = "0.24", features = ["extension-module"] }
```

## Implement `Splitter`

The splitter finds safe chunk boundaries. For CSV that means newline outside
quotes; for JSONL it is just newline; for JSON arrays it means brace balance.

```rust
use rypipe_core::Splitter;

pub struct CsvSplitter;

impl Splitter for CsvSplitter {
    fn find_split_points(&self, bytes: &[u8], max_chunks: usize) -> Vec<usize> {
        if max_chunks <= 1 || bytes.is_empty() {
            return vec![0, bytes.len()];
        }
        let mut points = vec![0];
        let mut in_quotes = false;
        for (i, &b) in bytes.iter().enumerate().skip(1) {
            if b == b'"' {
                in_quotes = !in_quotes;
            } else if b == b'\n' && !in_quotes && points.len() < max_chunks {
                points.push(i);
            }
        }
        if *points.last().unwrap() != bytes.len() {
            points.push(bytes.len());
        }
        points
    }

    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
        let newline_count = sample.iter().filter(|&&b| b == b'\n').count().max(1);
        (sample.len() / newline_count).max(1)
    }
}
```

Rules:

- The first point must be `0`; the last must be `bytes.len()`.
- Adjacent equal points produce empty ranges that the engine ignores.
- Each chunk must start at a valid row boundary.

## Implement `RecordParser`

```rust
use rypipe_core::{RecordParser, ColumnarSink, Value, Result};

pub struct CsvDecoder {
    header: Vec<String>,
}

impl RecordParser for CsvDecoder {
    fn validate(&self, bytes: &[u8]) -> Result<()> {
        simdutf8::basic::from_utf8(bytes)
            .map_err(|e| rypipe_core::Error::Utf8(e.to_string()))?;
        Ok(())
    }

    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;
            }
            sink.begin_row();
            for (col, value) in self.header.iter().zip(line.split(',')) {
                if sink.wants(col) {
                    sink.put_field(col, Value::Str(value));
                }
            }
            sink.end_row();
        }
        Ok(())
    }
}
```

Key points:

- Call `sink.wants(name)` before expensive extraction to skip dropped fields.
- Emit `Value::Str` for stringly formats; emit typed `Value` variants when the
  format has native numbers/booleans.
- Do not call `end_row()` for partial trailing rows; the engine will discard
  them.

## Run it with `Pipeline`

`Pipeline` is the recommended entry point. It handles file opening, plan
application, and all execution modes.

```rust
use rypipe_core::{ExecutionPlan, FieldType, Pipeline};

let pipeline = Pipeline::new(CsvSplitter, CsvDecoder {
    header: vec!["a".into(), "b".into()],
});

// Single-file parse.
let batch = pipeline.read_path("data.csv", false, false)?;

// Parallel parse.
let batches = pipeline.read_path_par("data.csv", 4, false, false)?;

// Bounded-memory streaming.
let batches = pipeline.read_path_stream(
    "huge.csv",
    rypipe_core::MemoryBudget::new(128 * 1024 * 1024),
    false,
)?;
```

## Pushdown plans with the builder API

```rust
use rypipe_core::{CompareOp, ExecutionPlan, FieldType};

let plan = ExecutionPlan::new()
    .rename("raw_amount", "amount")
    .drop("internal_id")
    .type_as("amount", FieldType::Float64)
    .type_as("quantity", FieldType::Int64)
    .dictionary("status")
    .filter_eq("status", "active")
    .schema_order(["quantity", "amount", "status"]);

let batch = pipeline.with_plan(plan).read_path("data.csv", false, false)?;
```

## Adding Python bindings

Your adapter package can expose its own Python module. Reuse `rypipe-python`
for the plan and export helpers:

```rust
use rypipe_python::{execution_plan_from_kwargs, record_batches_to_pyarrow_table};

#[pyfunction]
fn read_csv(
    py: Python<'_>,
    path: String,
    field_mapping: Option<HashMap<String, String>>,
    // ... other kwargs
) -> PyResult<PyObject> {
    let plan = execution_plan_from_kwargs(...)?;
    let batches = py.allow_threads(|| {
        // ... use Pipeline::read_path_par or BoundedExecutor
    })?;
    record_batches_to_pyarrow_table(py, &batches)
}
```

Then register the adapter with `rypipe` from Python:

```python
import rypipe

class CsvAdapter:
    def read(self, path, **kwargs):
        return _rypipe_csv.read_csv(path, **kwargs)

rypipe.register_adapter("csv", CsvAdapter(), extensions=[".csv"])
```

## Python `Adapter` subclass (pipeline API)

If your adapter returns a `pyarrow.Table`, the easiest way to expose a source
is to subclass `rypipe.Adapter` and implement ``read(self, path, **kwargs)``.
Plan kwargs are merged and passed through automatically::

```python
import rypipe
from rypipe import Adapter
import _rypipe_csv

class CsvAdapter(Adapter):
    def __init__(self, path, *, delimiter=",", **kwargs):
        super().__init__(path, **kwargs)
        self._delimiter = delimiter

    def read(self, path, **kwargs):
        return _rypipe_csv.read_csv(
            path, delimiter=self._delimiter, **kwargs
        )
```

For adapters that need full control, subclass `rypipe.Source` directly and
override `_read_arrow` and `_build_plan_kwargs`.

Users can now write::

```python
from rypipe import RenameFields, DropFields, FilterRows, CastTypes

src = CsvSource("data.csv")
df = (
    src
    | RenameFields({"old_name": "new_name"})
    | DropFields(["internal_id"])
    | FilterRows(field="status", op="==", value="active")
    | CastTypes({"amount": float})
).to_dataframe()
```

`RenameFields`, `DropFields`, `CastTypes`, and constant `FilterRows` expose
`_plan_kwargs()` so the pipeline fusion layer pushes them into `_read_arrow`.
Your Rust `read_csv` receives the merged kwargs and applies them in the parse
loop. Non-fusable stages (custom callables, compare filters, stateful transforms)
run over the returned table automatically.

## Testing an adapter

Recommended tests:

- Empty input.
- Single row.
- Multi-row with all field types.
- Rename/drop/type/filter pushdown via `ExecutionPlan`.
- Splitter invariants (monotonic points, no inverted ranges, coverage).
- Multi-chunk equivalence: parse whole file vs. split + merge.
- Partial trailing row discarded cleanly.
- `Pipeline::read_path`, `read_path_par`, and `read_path_stream` agree.

See the `rypipe-core` tests and the `bench_throughput` example for small
self-contained splitter/parser samples.

## See also

- [Rust API](./rust-api.md): `Pipeline`, `ExecutionPlan`, and `Value`.
- [Architecture](./architecture.md): how splitters, parsers, and the engine interact.
- [Python API](./python-api.md): registering adapters with the `rypipe` package.