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

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/adapter-design/
========================================================================

# Adapter design

A high-performance adapter does as little work as possible per record. This page covers the `Splitter` and `RecordParser` design patterns that keep rypipe fast.

## `Splitter` design

The splitter finds safe chunk boundaries for parallel parsing.

```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;
}
```

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.

A good splitter is cheap. It scans for boundaries with byte searches rather than parsing the whole chunk. For line-oriented formats, `memchr::memchr` finds newlines. For XML, `memchr::memmem` finds row tags.

## Finding split points with `memchr`

`memchr` is SIMD-accelerated on most platforms. A single `memmem` scan is much faster than iterating byte-by-byte.

Example from crxml:

```rust
use memchr;

let row_tag_count = memchr::memmem::find_iter(&sample[..sample_end], &self.row_tag).count();
```

For CSV, you also need to track quote state so a newline inside a quoted field does not become a false boundary.

## Handling comments, CDATA, and quoted fields

False split points corrupt chunks. A robust splitter skips regions that look like row boundaries but are not:

- XML comments: `<!-- ... -->`
- XML CDATA: `<![CDATA[ ... ]]>`
- CSV quoted fields: `"..."`
- JSON strings: `"..."`

In crxml, the splitter skips comments and CDATA while scanning for `<Row` tags. It also validates that a candidate tag is followed by whitespace, `>`, or `/` to avoid prefix collisions such as `<RowItem`.

## `RecordParser` tips

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

Best practices:

- Validate UTF-8 once per chunk with `simdutf8` in `validate`.
- In `parse_chunk`, walk events or lines and emit fields.
- Call `sink.wants(name)` before expensive extraction to skip dropped fields.
- Do not call `end_row()` for partial trailing rows; the engine discards them.

## Borrowing strings

If the input chunk is valid UTF-8, hand borrowed `&str` slices to the engine:

```rust
let text = std::str::from_utf8(bytes)?;
for line in text.lines() {
    sink.begin_row();
    sink.put_field("value", Value::Str(line));
    sink.end_row();
}
```

The engine copies the string into its arena only when necessary. Borrowing avoids per-field allocations in the parser.

## Sparse rows

If a field is missing, skip it entirely:

```rust
if let Some(value) = maybe_value {
    sink.put_field("status", Value::Str(value));
}
```

Do not emit `Value::Null` for every missing field. The engine null-fills missing columns at `end_row()`; emitting explicit nulls wastes work.

## Respecting `sink.wants`

`ColumnarSink::wants` lets the parser skip fields that will be dropped:

```rust
if sink.wants("internal_id") {
    sink.put_field("internal_id", Value::Str(extract_id(...)));
}
```

For expensive extractions (deep XML paths, regex captures), this is a major win. Always check `wants` before doing work that the engine will discard.

## `parse_tail` fallback

Chunks can start or end inside a row. A robust adapter has a fallback path that rescans from the nearest safe row start. crxml uses `parse_tail` to handle orphan close-tags at chunk boundaries without a serial pre-pass.

## Summary

- Split cheaply with `memchr`; defer full decoding.
- Skip comments, CDATA, quotes, and strings to avoid false boundaries.
- Borrow UTF-8 slices into the engine.
- Emit sparse rows and respect `sink.wants`.
- Handle trailing partial rows cleanly.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/anti-patterns/
========================================================================

# Anti-patterns

These patterns are common, legal, and expensive. Avoid them when throughput or memory matters.

## Iterating a table source row-by-row

```python
for row in pipeline:
    ...
```

This works, but it reconstructs Python dicts from the Arrow table. If the source is table-shaped and you need row access, consider using `to_arrow()` and `pyarrow` vectorized operations instead.

## Chaining Python callables

```python
result = (
    source
    | (lambda table: transform(table))
    | (lambda table: another_transform(table))
).to_arrow()
```

Each callable materializes a full Python object (usually a `pyarrow.Table` or list of dicts) and breaks fusion. Prefer fused stages or move the logic into Rust.

## Repeated `to_pandas` / `to_arrow`

```python
t1 = pipeline.to_pandas()
t2 = pipeline.to_arrow()
t3 = pipeline.to_pandas()
```

Each call re-runs the pipeline. Cache the table once and reuse it:

```python
table = pipeline.to_arrow()
t1 = table.to_pandas()
t2 = table
```

## Ignoring `plan_overrides`

```python
class MySource(Source):
    def _read_arrow(self, *, plan_overrides=None, **kwargs):
        return my_rust_read(self.path, **kwargs)  # plan_overrides lost!
```

If an adapter ignores `plan_overrides`, fused stages silently fall back to Python execution. Always forward `plan_overrides` to the Rust reader.

## Wrong engine choice

```python
source.read_stream(chunks=64)  # tiny file, over-parallelized
```

For small files, columnar mode is usually fastest. For huge files, stream mode keeps memory flat. Parallel mode only wins for large, CPU-bound, cached files.

## Using `auto_dict` in parallel mode for throughput

```python
source.read_par(auto_dict=True, chunks=32)
```

`auto_dict` forces the merge path in parallel mode. If throughput is the goal, use explicit `dictionary_columns` for only the columns that need it, or switch to columnar mode.

## Not declaring types for numeric filters

```python
FilterRows(field="amount", op=">", value="100.0")
```

Without `field_types={"amount": "float64"}`, the engine may store `amount` as a string and skip the vectorized compare filter. Declare the type so the filter runs in Arrow.

## Summary

- Cache tables; do not re-run pipelines.
- Forward `plan_overrides` in adapters.
- Keep Python callables out of the hot path.
- Match the engine mode to the file size and workload.
- Declare types for numeric filters.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/case-study-crxml/
========================================================================

# Case study: crxml

`crxml` is a high-throughput adapter for Crystal Reports XML exports. It is a concrete example of how the techniques from the other advanced pages combine to reach ~2.4 GB/s on a single workstation.

## What it parses

Crystal Reports exports tabular data inside XML elements such as:

```xml
<Field Name="amount"><Value>123.45</Value></Field>
<Text Name="status"><TextValue>active</TextValue></Text>
```

`crxml` reads these exports and turns them into Arrow tables or DataFrames. The speed is parser-bound; the `rypipe-core` engine keeps up without being the bottleneck.

## Architecture

```text
Crystal Reports XML file
    |
    v
CrystalXmlSplitter : finds row-tag boundaries
    |
    v
CrystalXmlDecoder : extracts fields from each row
    |
    v
rypipe-core engine : typed builders, filters, projection, Arrow export
    |
    v
pyarrow.Table / pandas.DataFrame
```

The Rust side lives in `crxml-core`. The Python side is a thin `CrystalXMLAdapter` that calls the Rust core and registers itself with `rypipe`.

## Techniques from this section

| Page | Technique used in crxml |
|------|-------------------------|
| [Adapter design](./adapter-design.md) | `memchr::memmem` splitter; skip comments/CDATA; validate tag boundaries. |
| [Adapter design](./adapter-design.md) | Borrowed-slice `quick_xml` reader; XML events point into the input buffer. |
| [Schema and types](./schema-and-types.md) | `field_types` casts strings to numbers during parse. |
| [Dictionary encoding](./dictionary-encoding.md) | `dictionary_columns` for low-cardinality string fields. |
| [Parallelism](./parallelism.md) | Parallel fast path when `auto_dict` and compare filters are off. |
| [Execution modes](./execution-modes.md) | `columnar`, `parallel`, and `stream` modes exposed through `rypipe`. |
| [I/O tuning](./io-tuning.md) | `mmap` with `prefault` for cached files; bounded streaming for huge files. |

## The splitter

`CrystalXmlSplitter` uses `memchr::memmem` to scan for the row tag. It is SIMD-accelerated on most platforms. It skips `<!-- ... -->` and `<![CDATA[ ... ]]>` regions so a `<Row` string inside them is not mistaken for a real row start. It also validates that a candidate tag is followed by whitespace, `>`, or `/` to avoid prefix collisions such as `<RowItem`.

## The decoder

`CrystalXmlDecoder` uses `quick_xml` in borrowed-slice mode. Events reference the input bytes directly instead of copying into a scratch buffer. For each row element it:

1. Emits row attributes as fields.
2. Walks child elements.
3. Recognizes `<Field>`, `<Text>`, and `<Section>` patterns.
4. Calls `sink.put_field(key, Value::Str(value))` so the engine builds typed columns.

The decoder also has a `parse_tail` fallback that rescans orphan close-tags at chunk boundaries, so chunked parsing stays correct without a serial pre-pass.

## Why it is fast

| Technique | Benefit |
|-----------|---------|
| Borrowed-slice `quick_xml` reader | XML events point into the input buffer; no per-event copy. |
| `memchr::memmem` row-tag scan | SIMD-accelerated boundary search for parallel chunks. |
| Skip-region handling | Comments/CDATA do not create false split points. |
| SIMD UTF-8 validation | `simdutf8` validates each chunk in bulk. |
| `rypipe-core` typed builders | Strings are copied into Arrow arrays only once, during parse. |
| Parallel fast path | When `auto_dict` and compare filters are off, chunks export independently. |

## Lessons for adapter authors

1. Specialize the parser. Generic line splitting is fine for engine benchmarks, but real throughput comes from a format-aware parser.
2. Find split points cheaply. A single `memmem` scan beats scanning byte-by-byte.
3. Handle boundary cases. Chunks can start or end inside a row; have a fallback path that rescans from the nearest safe row start.
4. Borrow strings into the engine. Pass `Value::Str(&str)` slices whenever the input is valid UTF-8.
5. Register with `rypipe`. A thin adapter class lets users call `rypipe.read()` while you keep the fast Rust core.

## Source

The full implementation is in the [crxml repository](https://github.com/emiliano-go/crxml), especially:

- `src/crxml_core/src/xml/splitter.rs`
- `src/crxml_core/src/xml/decoder.rs`
- `src/crxml/rypipe_adapter.py`

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/dictionary-encoding/
========================================================================

# Dictionary encoding

Arrow dictionaries store string values as integer indices into a separate value table. In `rypipe`, this can reduce memory 5-20x for low-cardinality string columns such as status codes, country codes, or enums. This page explains how dictionaries work in the engine, when they help, and when they force the merge path and hurt throughput.

## How Arrow dictionaries work in rypipe

`rypipe-core` stores string columns in a `StrColumn`: a contiguous byte arena plus `i32` offsets and a validity bitmap. When a column is dictionary-encoded, the engine instead builds:

- a `codes: Vec<Option<i32>>` array of indices;
- a `dict: Vec<String>` ordered list of distinct values;
- an `index: HashMap<String, i32>` lookup from value to index.

On Arrow export, these become a `DictionaryArray` with `Int32` indices and a `StringArray` dictionary. The layout is exactly what Arrow compute kernels expect, so downstream filters and group-by operations can use the encoded form directly.

## Explicit `dictionary_columns`

The safest way to use dictionary encoding is to declare it explicitly:

```python
source = MyAdapter(
    "data.log",
    schema=["id", "ts", "amount", "status"],
    field_types={"id": "int64", "amount": "float64"},
    dictionary_columns=["status"],
)
```

This tells the engine to build a dictionary column for `status` from the first row. There is no inference pass and no runtime heuristic cost.

In Rust:

```rust
let plan = ExecutionPlan::new()
    .dictionary("status");
```

## `auto_dict` heuristics

`auto_dict=True` asks the engine to guess which string columns should be dictionary-encoded. The heuristic has a small runtime cost: it tracks the number of distinct values and the total row count for each string column. When the ratio of distinct values to rows falls below a threshold, the column is upgraded to dictionary encoding at finish time.

Use `auto_dict=True` when:

- you do not know the schema or cardinality in advance;
- the file is small enough that the tracking cost is negligible;
- downstream operations benefit from dictionary form.

Use `auto_dict=False` when:

- throughput is the top priority;
- columns are high cardinality or already numeric;
- you are running parallel mode (see below).

## When dictionaries help memory

Dictionary encoding helps most when:

- the column has low cardinality (many repeated values);
- the strings are long relative to the index size;
- the column is used in filters, joins, or group-by operations that can work on integer codes.

Examples:

- HTTP status codes: ~10 distinct values, very short strings.
- Country codes: ~200 distinct values, short strings.
- Product categories: tens to thousands of distinct values, often repeated.

For very short strings (one or two characters), the memory savings are smaller because the string data is already small.

## When dictionaries force the merge path

In parallel mode, dictionary encoding forces the merge path. Each chunk builds its own local dictionary. Before export, the engine must merge all chunk dictionaries into a single global dictionary and remap codes. This has two consequences:

1. **Serial merge**: the merge step is not parallel, so it can become a bottleneck for many small chunks.
2. **Higher peak RSS**: all chunk builders must coexist until the merge finishes, and the global dictionary may be larger than any local one.

If you need both dictionaries and maximum throughput, consider:

- using columnar mode instead of parallel mode;
- declaring `dictionary_columns` explicitly so only those columns pay the merge cost;
- filtering after export instead of forcing a merge with compare filters.

## 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.

## Summary

- Use `dictionary_columns` for known low-cardinality strings; it is predictable and avoids heuristic cost.
- Use `auto_dict=True` only when cardinality is unknown and the file is small or not parallel.
- Remember that dictionaries force the parallel merge path; weigh memory savings against throughput loss.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/execution-modes/
========================================================================

# Execution modes

`rypipe` adapters can expose up to three execution strategies. Choosing the right one is usually the biggest single decision for memory and throughput.

| Mode | Best for | Memory | Parallelism | Output |
|------|----------|--------|-------------|--------|
| `stream` | Huge files, unknown schema, row-at-a-time consumers | bounded by batch size | single-threaded parse | iterator / batched record batches |
| `columnar` | Medium files that fit in RAM, table output | holds full table | single-threaded parse, vectorized builders | one `RecordBatch` |
| `parallel` | Large files that fit in RAM, many cores | holds full table | chunked multi-threaded parse | one `RecordBatch` or `Vec<RecordBatch>` |

`auto` lets the adapter pick. A common heuristic is: files under ~8 MiB use columnar; larger files use parallel when memory allows; otherwise stream. Adapters should document their own heuristic because format split boundaries affect chunk safety.

## Stream mode

Stream mode uses `BoundedExecutor`. It keeps a memory budget and parses the file in batches:

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 batches sized to fit the memory budget, capped at 256
   split points as an internal safeguard against pathological chunk counts.
5. Parses each batch into a `TableBuilder`, exports it to a `RecordBatch`, and resets the builder.
6. Returns a `Vec<RecordBatch>`; the caller concatenates or iterates.

Because the input buffer is dropped before the parse phase begins for bounded mode, mmap-backed pages are released before downstream work starts. This keeps peak memory close to the budget even for files much larger than RAM.

Use stream mode when:

- the file does not fit in RAM;
- the consumer is row-oriented or streaming (e.g., writing one row at a time);
- latency per batch matters more than total throughput;
- parallel merge overhead would dominate (very simple parsers).

## Columnar mode

Columnar mode parses the whole file in one thread and builds one `TableBuilder`. It is the simplest path and avoids all chunking and synchronization overhead. The full table stays in memory until export.

Use columnar mode when:

- the file fits comfortably in RAM;
- the parser is fast enough that parallel overhead would not pay off;
- you need one contiguous `RecordBatch` without a merge step;
- `auto_dict` or compare filters force a merge anyway, so parallelism adds overhead.

Columnar mode is often fastest for small files because there is no per-chunk setup and no rayon scheduling.

## Parallel mode

Parallel mode uses `ParallelExecutor`:

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. No serial merge happens.
5. Merge path: if `auto_dict` or a `Compare` filter is present, chunk builders are merged sequentially before export.

Use parallel mode when:

- the file fits in RAM or in the OS page cache;
- the parser is CPU-bound (heavy XML, complex field extraction, many columns);
- you can tolerate higher peak memory for shorter wall-clock time;
- `auto_dict` and compare filters are off, so the fast path applies.

## Auto engine selection

The Python `Adapter` layer usually exposes `engine="auto"`. The heuristic is format-specific, but a common default is:

```python
if file_size < 8 * 1024 * 1024:
    engine = "columnar"
elif memory_available > 4 * file_size:
    engine = "parallel"
else:
    engine = "stream"
```

Adapters should expose the engine choice explicitly because the best default depends on split safety, row size variance, and downstream use. A format with expensive per-chunk setup (for example, one that must scan for a global header) may prefer columnar for much larger files than a simple newline-delimited format.

## Trade-offs

| Concern | Prefer | Avoid | Why |
|---------|--------|-------|-----|
| Lowest memory | stream | parallel | Bounded batches keep peak RSS flat. |
| Lowest latency to first batch | stream | parallel | First batch is emitted before the whole file is read. |
| Highest throughput on large files | parallel | columnar | Many cores parse simultaneously. |
| Highest throughput on small files | columnar | parallel | Chunk overhead dominates. |
| Deterministic column order | any with `schema_order` | inference | Chunk merges rely on a common schema. |
| Low cardinality string compression | columnar or parallel with `dictionary_columns` | parallel with `auto_dict` | Auto-dict forces the merge path. |

## 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.

This means parallel mode can saturate CPU from Python without `multiprocessing`, provided the adapter is implemented in Rust and exports Arrow.

## Summary

- Use `stream` for huge files or row consumers.
- Use `columnar` for small-to-medium files and when merge is unavoidable.
- Use `parallel` for large cached files with a CPU-bound parser and no merge-forcing options.
- Expose `engine` explicitly and document the adapter-specific heuristic.

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

# Pushdown fusion

`rypipe` splits ingestion into two layers:

1. **Adapter layer**: reads bytes, splits them into records, and emits `Value` rows.
2. **Engine layer**: builds typed Arrow arrays, applies filters, and exports record batches.

Pushdown fusion is the process by which the Python `Pipeline` rewrites a chain of lightweight stages into a single `ExecutionPlan`. The engine then applies rename, drop, filter, and cast while it parses each row, instead of materializing a full table and running Python transforms afterward.

## A fused pipeline

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

source = MyAdapter("data.log")
result = (
    source
    | RenameFields({"old_name": "new_name"})
    | DropFields(["internal_id"])
    | FilterRows(field="status", op="==", value="active")
    | CastTypes({"amount": "float64"})
).to_arrow()
```

When the pipeline reaches `to_arrow()`, the stage list is collapsed into one plan. The Rust parser:

- renames `old_name` to `new_name` as fields arrive;
- skips `internal_id` entirely (it is not allocated);
- drops rows whose `status` is not `active` before they leave the builder;
- casts `amount` to `float64` once, during parse.

Rows that fail the filter are never materialized; dropped columns are never allocated; and casts happen once inside Rust instead of twice in Python and Rust.

## The `ExecutionPlan` fields

```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 (all kinds)
    pub schema_order: Vec<String>,                // output column order
    pub auto_dict: bool,                          // auto-dict upgrade
    pub dict_threshold: Option<f64>,              // auto-dict ratio (default 0.05)
    pub dict_max_size: Option<usize>,             // auto-dict max entries (default 256)
}
```

The plan is built by `_build_plan_kwargs()` on the Python side and consumed by `TableBuilder` on the Rust side.

## Field resolution order

Inside `TableBuilder`, every emitted field goes through this pipeline:

1. `field_map` renames the raw field name.
2. `drop_fields` checks the resolved name; if dropped, the field is ignored.
3. `field_types` / `dictionary_columns` chooses the storage type.
4. `filter` rejects rows during `end_row`.

This order matters. A filter runs on the resolved name, so it must be written in post-rename terms. A cast type is attached to the resolved name as well.

## What is fusable

Fusable stages implement `_plan_kwargs()` and merge cleanly into an `ExecutionPlan`:

| Stage | Plan field | Notes |
|-------|------------|-------|
| `RenameFields` | `field_map` | Multiple renames merge into one map. |
| `DropFields` | `drop_fields` | Merges as a set union. |
| `CastTypes` | `field_types` | Later casts overwrite earlier ones for the same field. |
| `FilterRows` predicate | `filter` | Constant (`field`/`op`/`value`) with `==`/`!=`, or column-to-column (`field_a`/`op`/`field_b`); both are evaluated per-row during parse. |
| `FilterRowsAny` / `FilterRowsAll` / `FilterRowsNot` | `filter` | `And`, `Or`, `Not` trees built from the same leaf shapes; evaluated per-row with short circuiting; fully fusable. |

`FilterRows` is fusable when it uses a constant predicate (`field`, `op`, `value`) with `==` or `!=`, or a column-to-column predicate (`field_a`, `op`, `field_b`). `FilterRowsAny`, `FilterRowsAll`, and `FilterRowsNot` are also fusable; they build `And`, `Or`, `Not` trees from the same leaves. All are evaluated per-row during parsing with native-typed comparison and numeric promotion; mismatched types or nulls fail the row, with `Not` flipping the result. Chaining `FilterRows` stages is an implicit `And` (see `plan_split`).

## What is not fusable

Non-fusable stages still work, but they run over the Arrow table after the engine finishes:

- Python callables (`lambda` or any callable stage).
- Stateful transforms such as window or aggregate stages.
- `FilterRows` wrapping a callable predicate (runtime-computed values).
- Custom stages that do not implement `_plan_kwargs()`.

When a non-fusable stage is present, the pipeline automatically falls back to a row stream or table transform path. The fusable prefix still runs in Rust; only the suffix runs in Python.

## Inspecting `plan_overrides` in an adapter

Adapters that subclass `rypipe.Source` receive fused plan kwargs through `_read_arrow`:

```python
class MySource(Source):
    def _read_arrow(self, *, plan_overrides=None, **kwargs):
        plan_overrides = plan_overrides or {}
        print(plan_overrides)
        # {
        #   "field_mapping": {"old_name": "new_name"},
        #   "drop_fields": ["internal_id"],
        #   "filter": {"field": "status", "op": "==", "value": "active"},
        #   "field_types": {"amount": "float64"},
        # }
        return my_rust_read(path=self.path, **plan_overrides, **kwargs)
```

The Rust side merges these kwargs into an `ExecutionPlan` with `execution_plan_from_kwargs` from `rypipe-python`, or constructs the plan manually:

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

let plan = ExecutionPlan::new()
    .rename("old_name", "new_name")
    .drop("internal_id")
    .type_as("amount", FieldType::Float64)
    .filter_eq("status", "active");
```

If an adapter ignores `plan_overrides`, fused stages silently fall back to Python execution over a full table. That is one of the most expensive anti-patterns.

## Order of operations across the pipeline

Fusable stages commute in the plan, but the engine applies them in a fixed order:

```
raw field name
    |
    v
rename (field_map)
    |
    v
drop check (drop_fields)
    |
    v
type selection (field_types / dictionary_columns)
    |
    v
per-row filter (Equal / NotEqual / Compare / And / Or / Not)
    |
    v
builder append (Vec plus map, single hash, dirty bitmask)
    |
    v
finish: sort by schema_order, auto-dict, Arrow export
```

Because drop happens before type selection, you cannot cast a dropped field.
Because every filter : including column-to-column comparisons : runs before the
row is committed, rejected rows consume no Arrow storage.

## When fusion does not help

Fusion is not free if the adapter cannot act on the plan. An adapter that always parses every field into Python objects and then builds Arrow will not benefit; the engine must receive fields through `put_field` and honor `wants()`. If the adapter is a thin wrapper around a library that returns full Python dicts, fusion only removes a small amount of Python overhead.

Fusion also cannot help when the workload is dominated by I/O. If the file is on a slow network share, reducing CPU work may not change wall-clock time. Profile first.

## Summary

- Fuse `RenameFields`, `DropFields`, `CastTypes`, and `FilterRows` predicates
  (constant and column-to-column) by implementing `_read_arrow(plan_overrides=...)`.
- Inspect `plan_overrides` to confirm that stages reach the Rust parser.
- Non-fusable stages run after the engine; keep them out of the hot path when throughput matters.

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

# Advanced rypipe

This section is for adapter authors and power users who want to understand why rypipe is fast and how to keep it fast. It assumes you have read the Python API, Rust API, and Architecture pages.

Each page takes one optimization topic and explains the mechanism, the tuning knobs, and the trade-offs.

## Roadmap

| Page | What you will learn |
|------|---------------------|
| [Fusion](./fusion.md) | How `RenameFields`, `DropFields`, `CastTypes`, and constant `FilterRows` are rewritten into a single `ExecutionPlan`; what is fusable and what falls back to Python. |
| [Execution modes](./execution-modes.md) | `stream`, `columnar`, `parallel`, and `auto`; when each wins on memory, latency, and throughput. |
| [Memory and chunking](./memory-and-chunking.md) | How `BoundedExecutor` enforces a memory budget; sizing chunks for files larger or smaller than RAM. |
| [Parallelism](./parallelism.md) | `rayon` internals; the `num_chunks` formula; why too many chunks hurt; measuring speedup. |
| [Dictionary encoding](./dictionary-encoding.md) | Arrow dictionaries in rypipe; `auto_dict` heuristics; the merge path vs the fast path; explicit `dictionary_columns`. |
| [Schema and types](./schema-and-types.md) | Using `schema_order` and `field_types` to skip inference passes, stabilize column order, and enable numeric compare filters. |
| [I/O tuning](./io-tuning.md) | `mmap` vs buffered reads; `prefault`; OS page cache; storage class considerations. |
| [Adapter design](./adapter-design.md) | Writing a fast `Splitter` and `RecordParser`; `memchr`; comments, CDATA, and quoted fields; borrowing strings; sparse rows; `sink.wants`. |
| [Profiling](./profiling.md) | Profiling with `perf`, `cargo flamegraph`, and the `bench_throughput` example; measuring RSS; separating Python and Rust time. |
| [Anti-patterns](./anti-patterns.md) | Common mistakes that silently remove fusion, increase memory, or waste CPU. |
| [Case study: crxml](./case-study-crxml.md) | How crxml reaches ~2.4 GB/s by combining the techniques from the other pages. |

## Quick checklist

- [ ] Provide `schema_order` and `field_types` when the schema is known.
- [ ] Use `dictionary_columns` for low-cardinality strings; be careful with `auto_dict` in parallel mode.
- [ ] Implement `_read_arrow(plan_overrides=...)` so fused stages stay in Rust.
- [ ] Pick `columnar` for tables that fit in RAM, `parallel` for large cached files, and `stream` for huge or row-oriented consumers.
- [ ] Tune `chunks = 4 * physical_cores` and measure with your data.
- [ ] Export Arrow from Rust and sink directly to Parquet when possible.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/io-tuning/
========================================================================

# I/O tuning

Parsing cannot outrun the I/O subsystem. `rypipe` supports memory-mapped and buffered input, plus prefault options. Choosing the right combination depends on file size, RAM, and storage class.

## `mmap` vs buffered reads

`use_mmap=True` (the default) maps the file into virtual memory. The kernel loads pages on demand and caches them in the OS page cache.

| Mode | Best for | Behavior |
|------|----------|----------|
| `use_mmap=True` | Files that fit in RAM or are expected to be cached | Kernel manages paging; parser sees a byte slice. |
| `use_mmap=False` | Large cold files or portability constraints | Reads the entire file into a `Vec<u8>`. |

For files that fit in RAM, mmap is usually fastest because it avoids an explicit copy into user space. For cold files larger than RAM, buffered reads may give smoother throughput because the parser avoids page-fault stalls.

## Prefault

`prefault=True` uses `MADV_WILLNEED` to fault the whole file up front. This is fastest when the file fits in RAM and you want to hide latency behind sequential reads. It can be harmful for files larger than RAM because it forces the kernel to read pages that may be evicted before use.

`prefault=False` uses `MADV_SEQUENTIAL` so the kernel can drop pages behind the reader. This is better for RSS-sensitive workloads and for streaming large files.

| 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>`. |

## OS page cache

The page cache is the biggest factor for repeated reads. If a file has been read recently, it is probably in cache, and mmap or buffered reads will be fast regardless of the underlying storage.

For one-off reads of large files, storage bandwidth is the limit. A modern NVMe SSD can sustain 3-7 GB/s sequential reads; a SATA SSD is closer to 500 MB/s; network storage varies widely.

## SSD vs NVMe vs network storage

| Storage | Typical sequential read | Implications |
|---------|------------------------|--------------|
| NVMe SSD | 3-7 GB/s | Parser can be the bottleneck; parallel mode helps. |
| SATA SSD | 400-600 MB/s | May be I/O-bound for simple formats; still fast enough for most XML/JSON. |
| Network (NFS/S3) | 50-500 MB/s | Latency and throughput vary; streaming may be safer than mmap. |
| Cold object storage | <100 MB/s | Consider downloading first or using buffered reads. |

On network storage, mmap can trigger many small page faults over a high-latency link. Buffered reads with a large readahead are usually smoother.

## Drop before parse

In bounded stream mode, the input buffer is dropped before the parse phase begins. This releases mapped pages before downstream work starts. Combined with `prefault=False`, this keeps peak memory close to the parse budget even when the file is much larger than RAM.

## Summary

- Use `mmap` + `prefault=True` for cached or RAM-resident files.
- Use `mmap` + `prefault=False` for large streaming files.
- Use buffered reads for network or portable deployments.
- Match the parser throughput to storage bandwidth; do not over-parallelize an I/O-bound workload.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/memory-and-chunking/
========================================================================

# Memory and chunking

`rypipe` tries to parse as fast as the hardware allows while staying inside a memory budget. Two knobs control that trade-off:

- `memory`: maximum bytes the parser should hold in flight. A string like `"512MiB"` is parsed into bytes.
- `chunks`: number of chunks for parallel mode. More chunks improve load balancing but increase scheduling overhead.

This page explains how `BoundedExecutor` enforces the budget and how to size chunks for files larger or smaller than RAM.

## How `BoundedExecutor` works

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

1. Opens the file via `InputBuffer`.
2. Estimates `bytes_per_row` from `Splitter::estimate_bytes_per_row`.
3. Computes `rows_per_batch` from the budget.
4. Splits the file into batches sized to fit the memory budget, capped at 256
   split points as an internal safeguard against pathological chunk counts.
5. Parses each batch into a `TableBuilder`, exports it to a `RecordBatch`, and resets the builder.
6. Returns a `Vec<RecordBatch>`; the caller concatenates.

The budget covers builder storage: string arenas, numeric buffers, and validity bitmaps. It does not include the input buffer, Arrow export buffers, or downstream pandas conversion. Set the budget lower than total RAM.

## Sizing the memory budget

A reasonable starting point for a workstation is 500 MiB. For a server with many concurrent parsers, divide available RAM by the expected concurrency. For embedded or container workloads, use 128 MiB or less.

The budget is a target, not a hard limit. Spikes can happen when:

- a batch contains an unusually wide row;
- a string column receives a very large value;
- the `bytes_per_row` estimate was wrong because of high variance.

If you see RSS overshoots, lower the budget and add more batches.

## Sizing chunks for parallel mode

Rule of thumb for parallel mode:

```
chunks = 4 * physical_cores
```

Finer chunks even out variable record parse times. Beyond 4-8x core count, synchronization overhead usually wins. Measure with your data; text-heavy formats benefit from fewer chunks because per-chunk setup dominates.

For a CPU-bound parser on many cores, start with 4x physical cores and increase until throughput flattens. For a memory-bandwidth-bound parser, fewer chunks may be better because each chunk touches the same memory hierarchy.

## Impact of row size variance

`BoundedExecutor` uses `bytes_per_row` to convert a byte budget into a row count. If rows vary in size, the row count can be wrong in either direction:

- Underestimate: a batch exceeds the budget and RSS spikes.
- Overestimate: batches are tiny and overhead rises.

High variance is common in:

- XML with mixed text and attribute payloads;
- JSON with nested arrays or large string fields;
- log files with variable field counts.

For these formats, prefer a smaller memory budget and more batches, or use stream mode with a conservative row estimate.

## Files larger than RAM

Stream mode is designed for this case. The input buffer is dropped before parsing, so mapped pages are released before downstream work starts. Each batch is parsed, exported, and discarded independently.

Tips:

- Use `prefault=False` so the kernel can drop pages behind the reader.
- Set `memory` to a fraction of RAM (for example, 25%).
- Avoid `auto_dict`; it forces a full table merge in parallel mode.
- Sink directly to Parquet or another stream-friendly format instead of building a pandas DataFrame.

## Files smaller than RAM

For small files, columnar mode is usually fastest. There is no chunk setup, no rayon scheduling, and no merge step. The entire file is parsed in one pass and exported once.

If the file is small but the parser is slow (for example, complex XML), parallel mode may still win despite overhead. Benchmark both.

## 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>>`.

## Summary

- Use `memory` to cap builder storage; leave headroom for export and downstream work.
- Start with `chunks = 4 * physical_cores` and tune by measurement.
- Reduce batch size when row size variance is high.
- Use stream mode for files larger than RAM; use columnar mode for small files.

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

# Parallelism

Parallel mode uses `rayon` to parse chunks concurrently. Understanding how `rayon` schedules work and how `num_chunks` maps to hardware helps avoid the common mistake of over-parallelizing.

## The `num_chunks` formula

A safe starting point is:

```
num_chunks = 4 * physical_cores
```

For a CPU-bound parser, this provides enough tasks to keep all cores busy even when chunks finish at different speeds. For a memory-bandwidth-bound parser, fewer chunks may be better because each chunk contends for the same DRAM channels and caches.

The built-in `bench_throughput` example is a simple TSV adapter. On a 12-core/24-thread Ryzen 9 5900X, parallel mode is slightly slower than single-threaded parse because the parser is so fast that chunk overhead dominates. Real adapters with heavier parsing usually see a win.

## How `rayon` schedules chunks

`ParallelExecutor` calls `rayon::par_iter` over the chunk ranges. `rayon` maintains a thread pool sized to the number of logical cores and uses work-stealing to balance load. Each chunk is parsed by one thread into its own `TableBuilder`.

Key properties:

- Tasks are not pinned to cores. The OS migrates them based on load.
- Work-stealing helps when chunks have variable cost.
- The global thread pool is shared with other `rayon` users in the same process.

## Hyperthreading

`rayon` uses logical cores by default. On a CPU with SMT (hyperthreading), two logical cores share execution units, L1, and L2. For memory-bandwidth-bound parsers, logical cores may not add much throughput. For CPU-bound parsers, they often add 10-30%.

If you want to exclude hyperthreads, set the `RAYON_NUM_THREADS` environment variable to the physical core count before starting Python:

```bash
export RAYON_NUM_THREADS=12  # physical cores only
python script.py
```

## NUMA and cache effects

On multi-socket or large NUMA machines, memory bandwidth and latency depend on which socket owns the buffer. `rayon` does not bind tasks to NUMA nodes, so a chunk parsed on socket 1 may read input allocated on socket 0.

For maximum throughput on NUMA hardware:

- allocate the input buffer on the same node that will do most of the parsing;
- pin the Python process to one socket if the file fits in one node's RAM;
- expect lower scaling when the file is larger than a single node's memory.

L3 cache size also matters. If the working set for one chunk fits in L3, scaling is good. If chunks are larger than L3, all cores contend for DRAM and speedup flattens.

## Why too many chunks hurt

More chunks are not always better. Each chunk pays fixed costs:

- `Splitter` validation and boundary search;
- `RecordParser` setup (readers, buffers, state);
- `TableBuilder` allocation and finish;
- Arrow export and, on the merge path, builder merging.

When the number of chunks grows, these fixed costs multiply. At some point the cost of starting a chunk exceeds the parsing work inside it. The result is lower throughput and higher memory use.

Too many chunks also increase peak RSS because each chunk holds its own builder until all chunks finish. On the merge path, all builders must coexist before the serial merge begins.

## Measuring speedup

Run the same parse at several chunk counts and plot throughput:

```bash
python benchmarks/bench_throughput.py --chunks 1 --output c1.json
python benchmarks/bench_throughput.py --chunks 4 --output c4.json
python benchmarks/bench_throughput.py --chunks 8 --output c8.json
python benchmarks/bench_throughput.py --chunks 16 --output c16.json
python benchmarks/bench_throughput.py --chunks 32 --output c32.json
```

Look for the elbow where adding chunks stops helping. Also measure RSS at each point; sometimes the fastest setting is not the most memory-efficient.

## Summary

- Start with `chunks = 4 * physical_cores`.
- Reduce chunks for simple, memory-bandwidth-bound parsers.
- Consider `RAYON_NUM_THREADS` to test physical-core-only behavior.
- Measure throughput and RSS; do not assume more parallelism is faster.

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

# Profiling

Optimization without measurement is guessing. This page describes how to profile rypipe pipelines and interpret the results.

## The `bench_throughput` example

`crates/rypipe-core/examples/bench_throughput.rs` is a self-contained benchmark. It uses a tiny inline TSV-like adapter so the result measures the engine, not an external parser.

Run it:

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

Or use the Python wrapper that writes JSON results:

```bash
python benchmarks/bench_throughput.py --output .benchmarks/rypipe.json
```

The output reports rows, time, rows per second, MB/s, and RSS. Use these to compare configurations.

## Release builds

Always profile a release build. Debug builds are 10-50x slower and the profile will be dominated by unrelated overhead.

```bash
cargo build --release -p rypipe-core
```

For symbols without full debug overhead, use the `profiling` profile if it exists:

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

## Profiling with `perf`

On Linux:

```bash
perf record -g cargo run --release -p rypipe-core --example bench_throughput
perf report -g 'graph,0.5,caller'
```

Look for time spent in:

- `find_split_points`: splitter is expensive.
- `parse_chunk`: parser is the bottleneck.
- `StrColumn::push` or builder append: string allocation/copy dominates.
- Arrow export or compute kernels: export is expensive.
- Python GIL-related functions: Python boundary is the bottleneck.

## Flamegraphs

`cargo flamegraph` produces an SVG flamegraph:

```bash
cargo install flamegraph
cargo flamegraph --release -p rypipe-core --example bench_throughput
```

Open `flamegraph.svg` in a browser. Wide bars are hot functions. Look for unexpected wide bars such as JSON serialization, Python dict construction, or allocations.

## Measuring RSS

Use `/usr/bin/time -v` on Linux:

```bash
/usr/bin/time -v cargo run --release -p rypipe-core --example bench_throughput
```

Look at `Maximum resident set size (kbytes)`. Compare this across engine modes and chunk counts.

In Python, you can sample RSS during a run with `psutil`:

```python
import psutil, time, os

proc = psutil.Process(os.getpid())
peak = 0
while running:
    peak = max(peak, proc.memory_info().rss)
    time.sleep(0.01)
print(f"peak RSS: {peak / 1024 / 1024:.1f} MiB")
```

## Separating Python and Rust time

If the pipeline includes Python stages, wrap the Rust parse in `py.allow_threads` (PyO3) so the GIL is released. Profile the Python side separately with `cProfile`:

```bash
python -m cProfile -o profile.stats script.py
python -c "import pstats; pstats.Stats('profile.stats').sort_stats('cumtime').print_stats(20)"
```

If most time is in `_rypipe` native code, optimize Rust. If most time is in Python callables, move work into fused stages or Rust.

## What to vary

When benchmarking, change one variable at a time:

- `chunks`: 1, 2, 4, 8, 16, 32.
- `memory`: 64 MiB, 256 MiB, 512 MiB, 1 GiB.
- `use_mmap` and `prefault`: all four combinations.
- `auto_dict`: on vs off.
- `field_types`: typed parse vs string inference.

Plot throughput vs RSS to find the Pareto frontier.

## Summary

- Use `bench_throughput` as a baseline.
- Profile release builds with `perf` or `cargo flamegraph`.
- Measure RSS separately; throughput is not the only metric.
- Separate Python time from Rust time before optimizing.

========================================================================
PAGE: https://rypipe.emiliano-go.com/advanced/schema-and-types/
========================================================================

# Schema and types

`rypipe` can infer column names and types, but inference passes cost time and memory. Providing `schema_order` and `field_types` up front avoids those passes, stabilizes column order, and enables numeric compare filters.

## Avoiding inference passes

Some formats need a discovery pass to infer column names. For example, an XML adapter may scan the file to find all field names before parsing. This doubles I/O work and delays the first row.

Provide `schema_order` when the columns are known:

```python
source = MyAdapter(
    "data.log",
    schema_order=["id", "ts", "amount", "status"],
)
```

With `schema_order`, the engine does not need to discover column names. It also sorts columns to this order at finish time, making output deterministic.

## Stable column order across chunks

In parallel mode, each chunk may encounter columns in a different order. Without a shared `schema_order`, the engine must reconcile column order at merge time. This adds a small per-chunk cost and can produce unexpected ordering when chunks disagree.

`schema_order` fixes the output order regardless of the order in which fields arrive.

## Casting during parse

`field_types` tells the engine which storage type to build for each column:

```python
source = MyAdapter(
    "data.log",
    field_types={
        "id": "int64",
        "ts": "string",
        "amount": "float64",
        "is_active": "bool",
    },
)
```

The engine builds the correct Arrow array from the first row. It does not store intermediate strings and recast later. This saves memory and CPU.

Supported types include:

| Type | Rust `FieldType` | Notes |
|------|------------------|-------|
| `string` / `str` | `FieldType::String` | Default for text data. |
| `int64` / `int` | `FieldType::Int64` | Parses integer strings during parse. |
| `float64` / `float` | `FieldType::Float64` | Parses float strings during parse. |
| `bool` / `boolean` | `FieldType::Boolean` | Parses common bool representations. |
| `dictionary` | `FieldType::Dictionary` | Dictionary encoding; equivalent to listing the column in `dictionary_columns`. |
| `date32` | `FieldType::Date32` | ISO dates (`YYYY-MM-DD`) stored as days since the Unix epoch. |
| `timestamp`, `timestamp[s]`, `timestamp[ms]`, `timestamp[us]`, `timestamp[ns]` | `FieldType::Timestamp(unit)` | ISO-8601 timestamps stored as integers in the given unit (default µs). |

`field_types={"status": "dictionary"}` and `dictionary_columns=["status"]` are
two spellings of the same storage decision; prefer `dictionary_columns` (or
`auto_dict`) so encoding choices stay separate from value types.

In Rust:

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

let plan = ExecutionPlan::new()
    .type_as("amount", FieldType::Float64)
    .type_as("quantity", FieldType::Int64);
```

## Numeric compare filters

Casting during parse is especially important for filters. When both sides of a
column-to-column comparison (`Compare`) are stored as `Int64` or `Float64`, the
engine compares them natively per-row during parsing with numeric promotion
(Int64 vs Float64 widens to f64) : no Python-level comparisons and no
post-assembly pass.

If the columns are left as strings, the comparison falls back to string
ordering, which is rarely what you want for numbers. Declare the types
explicitly to keep numeric comparisons native.

## Combining schema hints with fusion

`schema_order` and `field_types` are part of the `ExecutionPlan`. They merge cleanly with `RenameFields`, `DropFields`, and `FilterRows`:

```python
result = (
    MyAdapter("data.log", schema_order=["id", "amount"], field_types={"amount": "float64"})
    | RenameFields({"old_name": "amount"})
    | FilterRows(field="amount", op=">", value="100.0")
).to_arrow()
```

The filter runs on the renamed, typed column. Without `field_types`, the filter would fall back to Python or be skipped.

## Summary

- Provide `schema_order` to skip inference and stabilize output columns.
- Provide `field_types` to cast during parse and enable numeric Arrow filters.
- Combine both with fused stages for the fastest path through the engine.

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

# Streaming with constant memory

`rypipe` can stream arbitrarily large files with **constant memory** — even a 50 GB file on a 2 GB Raspberry Pi — by yielding `RecordBatch` objects one at a time and dropping each after the consumer returns.

## Bounded vs streaming

| Mode | API | Peak memory | When to use |
|---|---|---|---|
| `bounded` (collecting) | `BoundedExecutor::run` / `Pipeline::read_path_stream` → `Vec<RecordBatch>` | `budget + sum(batches)` — still grows with file size if you collect | `source.to_arrow()` with `memory="256MB"` for a single table |
| `streaming` (consuming) | `BoundedExecutor::run_stream` + `BatchConsumer` / `iter_record_batches` | `budget + one batch` — constant | `for batch in rypipe.iter_record_batches(..., memory="64KB"):` + `ParquetWriter` |

The engine already respects a `MemoryBudget` (`crates/rypipe-core/src/bounded.rs:19`) and `StreamingBatchIterator` (`crates/rypipe-core/src/streaming.rs:30`) reuses a single `Vec<u8>` chunk buffer (`chunk_buf.resize(chunk_len)`) and `TableBuilder::reset()` (`crates/rypipe-core/src/engine.rs:75`) to keep RSS at `budget + batch`.

## Memory guarantee

* **Rust-only:** `budget + batch + export buffer`. With `batch_size=1` and `memory="64KB"` and small rows (~1 KB for `crxml` `Details`), peak is a few tens of KB plus the `mmap` mapping (dropped after `plan_chunks` `bounded.rs:52`). This is the **64 KB** target in the spec.
* **Python:** `pyarrow.RecordBatch` + interpreter overhead make true 64 KB impossible, but `iter_record_batches` is still bounded — `benchmarks/bench_extended.py` `bounded 64MB` 494 MB/s vs `bounded 64KB` 607 MB/s on 1 GB, and `50 GB` extrapolates to `~84s` at `64KB` vs `18s` `par32` full-RAM.

## Rust API

```rust
use rypipe_core::{BatchConsumer, BoundedExecutor, MemoryBudget, StreamingBatchIterator};
use arrow::record_batch::RecordBatch;

struct ParquetConsumer { writer: arrow::ipc::FileWriter<File> }
impl BatchConsumer for ParquetConsumer {
    fn consume(&mut self, batch: RecordBatch) -> rypipe_core::Result<()> {
        self.writer.write(&batch).map_err(|e| rypipe_core::Error::Arrow(Box::new(e)))?;
        Ok(())
    }
}

let budget = MemoryBudget::new(64 * 1024);
let splitter = CrystalXmlSplitter::with_row_tag("Details");
let parser = CrystalXmlDecoder::with_row_tag("Details");
let plan = ExecutionPlan::new();
let executor = BoundedExecutor::new(budget);
let mut consumer = ParquetConsumer { writer };
executor.run_stream(path, &splitter, parser, plan, false, &mut consumer)?;

// Or pull-based:
let iter = StreamingBatchIterator::new(path.to_path_buf(), splitter, parser, plan, budget, false);
for batch in iter {
    let batch = batch?;
    // handle batch
}
```

`Pipeline` convenience: `pipeline.read_bytes_stream_consumer(&bytes, budget, &mut consumer)` and `pipeline.read_path_stream_consumer(path, budget, false, &mut consumer)` (`crates/rypipe-core/src/pipeline.rs`).

## Python API

```python
import pyarrow.parquet as pq, rypipe

# High-level: rypipe handles adapter lookup + streaming
writer = pq.ParquetWriter("out.parquet", schema)
for batch in rypipe.iter_record_batches("50GB.xml", format="crxml", memory="64KB", batch_size=1, row_tag="Details"):
    writer.write_batch(batch)
writer.close()

# Direct via crxml
from crxml import CrystalXMLSource
src = CrystalXMLSource("50GB.xml", row_tag="Details")
for batch in src.iter_record_batches(memory="64KB"):
    writer.write_batch(batch)

# Pipeline
from crxml import DropFields, FilterRows
pipe = CrystalXMLSource("50GB.xml", row_tag="Details") | DropFields(["Field22"]) | FilterRows(field="Level", op="==", value="3")
for batch in pipe.iter_record_batches(memory="256MB"):
    writer.write_batch(batch)
```

`batch_size` overrides the budget-derived `rows_per_batch = budget / estimate_bytes_per_row` (`crates/rypipe-core/src/splitter.rs:41`). Default derives from `memory`; pass `batch_size=1` for minimal per-batch memory.

## When streaming falls back

`Pipeline.iter_record_batches` checks `plan_split` `rypipe/fusion.py:130`; if `remaining` non-fusable stages exist, it falls back to `iter_arrow_batches` (materialized). The same happens for `Source.iter_record_batches` when `_iter_record_batches_stream` is not implemented — it yields `to_arrow().to_batches()`. For constant memory, keep stages fusable (`RenameFields`, `DropFields`, `CastTypes`, `FilterRows` constant).

## Testing

Parse 1 GB `test_1gb.xml` `memory="64KB"` `batch_size=58` vs columnar (`bench_extended.py`): both 926,746 rows, `1.69s 607 MB/s` vs `1.74s 588 MB/s` within 10%. Assert `RSS < budget*2` via `resource.getrusage`.

## See also

* `crates/rypipe-core/src/consumer.rs` `BatchConsumer`
* `crates/rypipe-core/src/streaming.rs` `StreamingBatchIterator` (`sync_channel(1)` backpressure, `allow_threads` in `rypipe-python`)
* `crates/rypipe-core/src/bounded.rs:96` buffer reuse
* `docs/architecture/streaming.md`

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

# Columnar storage

This page documents `crates/rypipe-core/src/columnar.rs` in full. It is the storage layer that makes `TableBuilder` fast and Arrow export cheap.

## StrColumn

```rust
pub(crate) struct StrColumn {
    data: Vec<u8>,
    offsets: Vec<i32>,
    validity: Vec<bool>,
}
```

This is exactly the Arrow `StringArray` layout (offsets plus bytes plus null bitmap) without per cell `String` allocation.

* `data: Vec<u8>` is one contiguous arena. Every string's bytes are appended sequentially.

* `offsets: Vec<i32>` has `len + 1` entries. `offsets[i] .. offsets[i+1]` is the byte range for value `i`. It is initialized with `[0]` so `push` can compute the next offset as `data.len()`.

* `validity: Vec<bool>` marks null vs present. `true` means present; `false` means null (the buffer contains no bytes for that slot, but offsets still advance by 0).

Operations:

* `with_capacity(cap)` preallocates `offsets` with `cap + 1` and `data` with `cap * 16` (heuristic 16 bytes per string). This matches `TableBuilder::estimated_rows`.

* `push(v: Option<&str>)` extends `data` if `Some`, pushes `data.len()` to `offsets`, and pushes `is_some` to `validity`. No allocation per cell beyond the arena growth.

* `pop` undoes the last push: pops `validity`, pops `offsets`, truncates `data` to the last offset.

* `len` is `validity.len()`.

* `get(i: usize) -> Option<&str>` checks validity, slices `data[offsets[i] .. offsets[i+1]]`, and does `from_utf8` (the slice is known UTF-8 from `validate`, but the check is kept for safety).

* `append(&mut self, other: &StrColumn)` merges another column by base shifting offsets: `base = self.data.len() as i32`, then `self.offsets.extend(other.offsets[1..].iter().map(|o| o + base))`. This is O(n) in offsets, not in bytes.

* `to_arrow() -> Result<ArrayRef>` builds an Arrow `StringArray` by wrapping the three buffers with `OffsetBuffer`, `Buffer`, and `NullBuffer`. When all validity are true, `nulls` is `None`. This is a block copy of two buffers, not per cell.

## ColumnBuilder

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

Each variant stores a dense `Vec<Option<T>>` (or `StrColumn` for strings). There is exactly one builder per column, created by `ExecutionPlan::column_type` at first use.

* `String` is `StrColumn`.
* `Int64`, `Float64`, `Boolean` are `Vec<Option<T>>` with `lexical::parse` for string inputs.
* `Date32(Vec<Option<i32>>)` stores days since epoch. Parsing uses `chrono::NaiveDate::parse_from_str("%Y-%m-%d")`.
* `Timestamp(TimeUnit, Vec<Option<i64>>)` stores raw integers in the column's `TimeUnit` (Second, Millisecond, Microsecond, Nanosecond). Parsing tries `"%Y-%m-%dT%H:%M:%S%.f"`, then `" %H:%M:%S%.f"`, then bare `"%Y-%m-%d"` as midnight. Timezone handling is left to adapters that emit `Value::Timestamp` directly.
* `Dictionary { codes, dict, index }` stores `codes: Vec<Option<i32>>` plus `dict: Vec<String>` (id to string) and `index: HashMap<String,i32>` (string to id). This is the write path for `FieldType::Dictionary` and for auto dict upgrade.

Variant keys for unification (10 strings):

* `string`, `int64`, `float64`, `boolean`, `date32`, `timestamp[s]`, `timestamp[ms]`, `timestamp[us]`, `timestamp[ns]`, `dictionary`

`variant_key(&self) -> &'static str` returns the key. Timestamp units are distinguished so merging `timestamp[s]` with `timestamp[ms]` is an error rather than silent promotion.

## Push paths

`push_value(&mut self, value: Value<'_>)` is called for every field of every row. It handles typed `Value` variants:

* `Value::Null` becomes `None`.
* `Value::Str(s)` calls `push_str(Some(s))`, which parses according to column type: `lexical::parse` for numbers, `parse::<bool>` for booleans, `parse_date32`/`parse_timestamp` for temporals. Unparseable becomes `None`.
* `Value::Int64(i)` into `Int64` is native, into `Float64` widens `i as f64`, into `String` stringifies, into `Dictionary` encodes via `dict_code`.
* Similarly for `Float64`, `Bool`, `Date32`, `Timestamp`. Cross type mismatches (for example `Bool` into `Int64`) become `None`.

`push(&mut self, value: Option<String>)` and `push_str(&mut self, value: Option<&str>)` are the string entry points. `push_str` avoids allocation for typed columns (it parses and discards the string). Both handle `Dictionary` by calling `dict_code`.

`dict_code(dict: &mut Vec<String>, index: &mut HashMap<String,i32>, v: &str) -> i32` does `if let Some(&code) = index.get(v) { return code }` else `dict.push(v.to_owned())` and insert. Average O(1).

## Auto dictionary

`try_upgrade_to_dict(&mut self, min_rows: usize, max_ratio: f64, max_size: usize)` upgrades a `String` builder to `Dictionary` when cardinality is low. Steps:

1. Only `String` builders; others are no ops.
2. If `len < min_rows` (512 in `TableBuilder::auto_dict_upgrade`), leave as `String`.
3. Count distinct via `FxHashSet<&str>` over `iter().flatten()` (skipping nulls).
4. Compute cap: `ratio_cap = ((len as f64 * max_ratio) as usize).max(16).min(max_size)`. Floor of 16 lets tiny columns upgrade; cap respects `dict_threshold` (default 0.05) and `dict_max_size` (default 256).
5. If distinct > cap, leave as `String`.
6. Otherwise build `dict`, `index`, `codes` from the old `StrColumn` via `dict_code`.

Called after each chunk parse when `plan.auto_dict` is true, and after merge via `TableBuilder::auto_dict_upgrade` which respects `plan.dict_threshold` and `plan.dict_max_size`.

## Merging and promotion

`extend_owned(&mut self, other: ColumnBuilder) -> Result<()>` merges `other` into `self` by consuming `other`. Both must be the same variant (after promotion). Cases:

* `String` via `StrColumn::append` (base shift)
* `Int64`/`Float64`/`Boolean`/`Date32`/`Timestamp` via `Vec::append`
* `Timestamp` checks `unit_a == unit_b` else `Error::Merge`
* `Dictionary` remaps `b`'s dictionary into `a`'s via `dict_code` per value, then translates codes via `remap[idx]`

`unify_variants(a: &str, b: &str) -> Option<&'static str>` reconciles two variant keys:

* same → same
* `int64` plus `float64` → `float64`
* `string` plus `dictionary` → `dictionary`
* otherwise `None` (irreconcilable)

`promote_to_variant(&mut self, target: &'static str) -> Result<()>` mutates in place:

* `Int64` to `Float64` via `std::mem::take` then `map(|o| o.map(|n| n as f64))`
* `String` to `Dictionary` via taking the `StrColumn`, building `dict`/`index`/`codes`
* Same key is a no op
* Any other target returns `Error::Merge`

Used in `merge::extend` and `merge::engines_to_record_batches` before `extend_owned`.

## Arrow export

`arrow_datatype(&self) -> DataType` maps each variant to Arrow type: `Utf8`, `Int64`, `Float64`, `Boolean`, `Date32`, `Timestamp(unit, None)`, `Dictionary(Int32, Utf8)`.

`to_arrow_array(&self) -> Result<ArrayRef>` builds the native array:

* `String` via `StrColumn::to_arrow`
* `Int64`/`Float64`/`Boolean`/`Date32` via `iter().copied().collect::<Array>()`
* `Timestamp(unit, v)` via `collect::<PrimitiveArray<TimestampXType>>` per unit
* `Dictionary` via `Int32Array` keys plus `StringArray` values into `DictionaryArray::<Int32Type>::try_new`

## Typed value view

`TypedValue<'a>` is a borrowed view for filter evaluation:

```rust
pub(crate) enum TypedValue<'a> { Str(&'a str), Int64(i64), Float64(f64), Bool(bool), Date32(i32), Timestamp(i64) }
```

`get_typed_value(&self, index: usize) -> Option<TypedValue<'_>>` borrows directly from storage (dictionary decodes to `Str` via dict lookup). `get_filter_value` formats as `String` for `Equal`/`NotEqual` (dates via `format_date32`, timestamps via `format_timestamp`).

This enum lets `FilterPredicate::Compare` run natively per row with numeric promotion, without allocation.

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

# Data flow

This page shows how bytes move through the system in each execution mode. All modes share the same `Splitter` plus `RecordParser` plus `ExecutionPlan`; only the driver differs.

Legend for every diagram on this page: `(ADAPTER BOUND)` is code you write in the adapter crate (format specific). `(CORE)` is code in `rypipe` crates (format agnostic, reused).

## Single thread

```
(CORE) Pipeline::read_bytes(bytes)  or  Pipeline::read_path(path) -> InputBuffer::open -> as_slice
    |
    v
(CORE) TableBuilder::with_plan(cap, plan)     cap = bytes.len() / 512  (min 64)
    |
    v
(ADAPTER BOUND) parser.validate(bytes)?                 (CORE) simdutf8 check, called once per chunk; adapter decides what to validate
(ADAPTER BOUND) parser.parse_chunk(bytes, &mut sink)    (CORE) sink is TableBuilder (ColumnarSink)
         loop lines -> (CORE) begin_row (no op) -> (ADAPTER) put_field(s) -> (CORE) push_field_resolved plus dirty -> (CORE) end_row -> (CORE) finish_row per row
    |
    v
(CORE) TableBuilder::finish() -> RecordBatch   normalize, auto_dict_upgrade, sort_columns, to_arrow_array per column
```

One `RecordBatch` is returned. `InputBuffer::Owned` holds the bytes for the duration of the parse; there is no chunking. `ExecutionPlan` is applied per row in `finish_row` (`filter.check`).

## Parallel

```
(CORE) Pipeline::read_bytes_par(bytes, num_chunks)  or  read_path_par(path, num_chunks)
    |
    v
(ADAPTER BOUND) splitter.find_split_points(bytes, num_chunks) -> Vec<usize>  sorted, 0 and len included  (CORE) helper split_points_to_ranges
split_points_to_ranges(&points, len) -> Vec<Range>  (CORE)
    |
    v
(CORE) rayon::into_par_iter over Ranges  (thread pool, work stealing)
    each range -> (CORE) TableBuilder::with_plan(est, plan.clone())
                (ADAPTER BOUND) parser.validate(&bytes[range])?
                (ADAPTER BOUND) parser.parse_chunk(&bytes[range], &mut sink)?  (CORE) sink is TableBuilder
                Ok(sink)
    (CORE) catch_unwind per chunk -> Error::Merge("worker panicked ...") on panic
    collect::<Result<Vec<TableBuilder>>>()
    |
    v
(CORE) if !plan.auto_dict && schemas_consistent(&engines) {
    engines_to_record_batches(engines, &plan)   // fast path (CORE) (parallel Arrow build, unified schema, null_array for missing)
} else {
    merged = TableBuilder::with_plan(engines.len().max(64)*512, plan)  (CORE)
    for e in engines { merged.extend(e)? }      // merge path (CORE) (sequential extend with promotion)
    batch = merged.finish()  (CORE)
    if filter.is_some() { apply_compare_filter(batch, filter) }  (CORE) (pure Compare and And only)
    vec![batch]
}
```

Fast path (`engines_to_record_batches`) keeps one batch per chunk (chunked columns, no copy). It unifies schema via `unify_variants` and `promote_to_variant` so all batches share one `Schema`; missing columns are `null_array`. `rayon::par_iter` builds arrays in parallel. Merge path (`extend` loop) returns a single merged batch and handles `auto_dict` visibility (full cardinality) and irreconcilable type errors with `Error::Merge` naming the column.

Schemas consistent check (`parallel.rs:101`) builds `base: HashMap<&str,&str>` from `first.field_index` and `variant_key`, then verifies every other engine's `field_index` has the same key. Fast path is chosen only when `!auto_dict` and consistent.

## Bounded

```
(CORE) Pipeline::read_bytes_stream(bytes, budget)  or  read_path_stream(path, budget, prefault)
    |
    v
(CORE) BoundedExecutor::plan_chunks(bytes, splitter)   // splitter is (ADAPTER BOUND), the rest is (CORE)
    (ADAPTER BOUND) bytes_per_row = estimate_bytes_per_row(bytes).max(1)  (CORE) arithmetic
    total_rows_est = bytes.len() / bytes_per_row  (CORE)
    rows_per_batch = (budget.bytes() / bytes_per_row).max(1).min(total_rows_est.max(1))  (CORE)
    num_batches = (total_rows_est / rows_per_batch).max(1)  (CORE)
    (ADAPTER BOUND) split_points = splitter.find_split_points(bytes, num_batches.min(256))  (CORE) caps at 256
    chunks = split_points_to_ranges  (CORE)
    |
    v
(CORE) batch_engine = TableBuilder::with_plan(bytes_per_row.max(64), plan)
rows_in_batch = 0
for chunk in &chunks {
    chunk_bytes = &bytes[chunk.start..chunk.end]          // run_bytes path (CORE) slicing
    // or for run(path) with Mmap: (CORE) seek plus read_exact into Vec<u8> (file IO)
    (CORE) chunk_engine = TableBuilder::with_plan(chunk.len()/512, plan.clone())
    (ADAPTER BOUND) parser.validate(chunk_bytes)?; (ADAPTER BOUND) parse_chunk(chunk_bytes, &mut chunk_engine)?  (CORE) sink is TableBuilder
    (CORE) batch_engine.extend(chunk_engine)?          // single hash for new columns, Vec append otherwise
    rows_in_batch += chunk_rows
    if rows_in_batch >= rows_per_batch {
        batches.push(batch_engine.finish()?); batch_engine.reset(); rows_in_batch = 0  (CORE)
    }
}
if batch_engine.num_rows() > 0 { batches.push(batch_engine.finish()?) }  (CORE)
apply_plan_filter(&mut batches, &plan)  // pure Compare/And reapplication only (CORE)
```

`run_bytes` slices directly from `bytes` (used for decompressed buffers and for `Pipeline::read_bytes_stream`). `run` opens `InputBuffer`; if `Mmap` it drops the mapping after `plan_chunks` and reopens the file for `seek` plus `read_exact` per chunk (bounded RSS); if `Owned` it delegates to `run_bytes`. `MAX_SPLIT_CHUNKS = 256` caps split points.

## Input buffering  (CORE)

```
(CORE) InputBuffer::open(path, use_mmap, prefault)  (rypipe-core/src/input.rs)
    |
    +-- (CORE) detect_compression(path) reads 4 bytes
    |       1f 8b -> gzip (if feature gzip) (CORE) flate2, 28 b5 2f fd -> zstd (CORE) zstd crate, 04 22 4d 18 -> lz4 frame (CORE) lz4_flex
    |       if Some -> Owned(decompress(path, codec)?)  // read_to_end (CORE)
    |
    +-- else if cfg(mmap) && use_mmap -> (CORE) Mmap(MmapHandle::new(file, prefault)?) with WillNeed or Sequential
    +-- else -> (CORE) Owned(fs::read(path)?)
    |
    v
(CORE) input.as_slice() -> &[u8]  // passed to Pipeline::read_bytes variants or to BoundedExecutor
```

Note: adapter never touches `InputBuffer` directly; `Pipeline` does. Decompression is transparent (`Owned` served from memory) across all modes. `Mmap` is only for uncompressed files when the `mmap` feature is enabled and `use_mmap` is true.

Decompressed bytes are served from memory for all modes. `Pipeline::read_path` and `read_path_par` simply do `input.as_slice()` and call the bytes variants, so compression is transparent.

## Column lifecycle inside a row  (CORE) with (ADAPTER BOUND) events

```
(CORE) begin_row (no op)  // row boundaries tracked by row_count plus row_dirty
  (ADAPTER BOUND) put_field(k, v) -> (CORE) resolve(k) (ExecutionPlan::resolve_field, one hash) -> (CORE) ensure_column_idx (single hash for field_index plus Vec push if new) -> (CORE) row_dirty[idx]=true -> (CORE) last_write_wins check (if len > row_count { pop }) -> (CORE) push_value (lexical parse or typed)
  (ADAPTER BOUND) put_field(k, v) duplicate in same row -> (CORE) pop previous value for this row (len > row_count) then push new, dirty stays true
  ... (ADAPTER BOUND) may call put_field_resolved(r, v) after resolve(r) to avoid second hash (see Decoder)
end_row -> (CORE) finish_row
    for (i,b) in columns.iter_mut().enumerate() {  (CORE)
        if !row_dirty[i] { b.push(None) }  // null fill only missing (CORE)
        else { row_dirty[i]=false }       // clear for next row (CORE)
    }
    if (CORE) filter.check(...) false { for b in columns { b.pop() } return }  // per row And/Or/Not with short circuit (CORE)
    row_count += 1  (CORE)
```

`row_dirty` is `Vec<bool>` with the same length as `columns` (kept in sync in `ensure_column_idx` plus `take_column` plus `extend`). See [Engine](./engine.md) and [Optimizations](./optimizations.md) for why this saves 80 percent of `push(None)` calls.

Adapter code never touches `row_dirty`; it only emits `put_field` events. The engine owns the dirty tracking.

`row_dirty` avoids a `while b.len() < target` check for touched columns and avoids pushing `None` for them. See [Engine](./engine.md) and [Optimizations](./optimizations.md).

## Error handling

All drivers propagate `Result` via `?`. Panics in parallel workers are caught with `catch_unwind` and turned into `Error::Merge`. Type mismatches in `extend` and `engines_to_record_batches` become `Error::Merge` with column name and hint to provide `field_types`. UTF-8 failures become `Error::Utf8`, I/O into `Error::Io`, Arrow construction into `Error::Arrow`, plan problems into `Error::Plan`. Python maps these to `ParseError`, `PlanError`, `MergeError`.

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

# Decoder API

`crates/rypipe-core/src/decoder.rs` (63 lines) defines the boundary between format specific and format agnostic code. Adapters implement two traits; the engine implements the third.

## 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 where the input may be split. The first should be `0` and the last should be `bytes.len()`. Adjacent offsets produce one `Range`. The helper `split_points_to_ranges(&points, len) -> Vec<Range<usize>>` turns points into non empty ranges via `windows(2)` and `filter_map(|w| if start < end { Some(start..end) } else { None })`.

* `estimate_bytes_per_row` is used by `BoundedExecutor` to size batches (`rows_per_batch = budget / bytes_per_row`). It is called once on the whole input (or on `bytes` for `run_bytes`).

Rules for a correct splitter: points are sorted, start at a valid row boundary, and point at the first byte of a record, not at the delimiter itself (see `docs/writing-adapters.md` for the CSV example with `i + 1` after `\n`).

## 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` is called once per chunk before `parse_chunk`. For stringly formats it is `simdutf8::basic::from_utf8(bytes)?` (SIMD). For typed formats it may be a no op.

* `parse_chunk` turns a chunk into `begin_row`, `put_field`, `end_row` calls on the sink. It must not call `end_row` for a partial trailing row; the engine discards it via `normalize`. It should handle sparse rows (skip missing) and last write wins is handled by the sink, not the parser.

Parsers never see `ExecutionPlan`. They emit raw field names as they appear in the format. The sink resolves them.

## 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 resolve<'a>(&'a self, name: &'a str) -> Option<&'a str> { Some(name) }
    fn put_field_resolved(&mut self, resolved_name: &str, value: Value<'_>) { self.put_field(resolved_name, value) }
    fn finish(&mut self) -> Result<RecordBatch>;
}
```

This is the event sink that decoders drive.

* `begin_row` and `end_row` bracket a row. `TableBuilder` uses `row_count` plus `row_dirty` to track the row, so `begin_row` is a no op.

* `put_field(&mut self, name: &str, value: Value<'_>)` resolves `name` via `ExecutionPlan::resolve_field` (rename then drop) and stores the value. If `resolve` returns `None` (dropped), it returns immediately.

* `wants(&self, name: &str) -> bool` is the hint: return false to signal the engine will drop this field. Default is true. Adapters that do expensive extraction (entity unescaping, base64, decompression) call `if sink.wants(col) { /* decode */ sink.put_field(col, val) }` to skip work.

* `resolve<'a>(&'a self, name: &'a str) -> Option<&'a str>` is the single lookup version. Default returns `Some(name)` (keep as is). `TableBuilder` overrides to `self.plan.resolve_field(name)` which returns `Some(resolved)` or `None` for dropped, borrowing from `field_map` where possible. This lets adapters do `if let Some(r) = sink.resolve(k) { /* expensive decode */ sink.put_field_resolved(r, v) }` with one hash instead of two (`wants` plus `put_field`).

* `put_field_resolved(&mut self, resolved_name: &str, value: Value<'_>)` pushes a field that is already resolved. Default delegates to `put_field` (which will resolve again). `TableBuilder` overrides to `push_field_resolved` which calls `ensure_column_idx` directly and sets `row_dirty[idx] = true` without re hashing `field_map`/`drop_fields`. This is the fast path for adapters that already called `resolve`.

* `finish(&mut self) -> Result<RecordBatch>` finalizes the sink. For `TableBuilder` it does `normalize`, early `new_empty` if no columns, `auto_dict_upgrade`, `sort_columns`, and builds `Schema` plus arrays.

### Why two APIs for the same thing

`wants` plus `put_field` is backward compatible and simple for stringly adapters (CSV header loop). `resolve` plus `put_field_resolved` is the same semantics with one hash instead of two, and it avoids the extra `String` allocation in `push_field` when `field_map` is non empty (`owned = n.to_owned()`). The Python fusion layer and `merge.rs` already use the single lookup Vec path; adapters can choose either pair and the engine guarantees the same result. Tests in `tests/data_integrity_test.rs` (`resolve_put_field_resolved_identical_to_put_field`) assert bit identical batches across `LineParser` vs `LineParserResolved` for 1000 rows with rename, drop, filter, and typed columns across single, parallel, and bounded modes.

## Value

`crates/rypipe-core/src/value.rs` (`Value<'a>`):

```rust
pub enum Value<'a> {
    Str(&'a str),
    Int64(i64),
    Float64(f64),
    Bool(bool),
    Date32(i32),
    Timestamp(i64),
    Null,
}
```

`Str(&str)` borrows from the input buffer (zero allocation for stringly formats). Typed variants let JSON adapters emit native numbers without string round tripping. `ColumnBuilder::push_value` handles cross type coercion (for example `Int64` into `Float64` widens, into `String` stringifies).

## Typical adapter loop

```rust
fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
    let text = std::str::from_utf8(bytes).map_err(|e| crate::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(',')) {
            // Simple path:
            // if sink.wants(col) { sink.put_field(col, Value::Str(value)); }

            // Fast path when extraction is expensive:
            if let Some(resolved) = sink.resolve(col) {
                // ... heavy decode of `value` ...
                sink.put_field_resolved(resolved, Value::Str(value));
            }
        }
        sink.end_row();
    }
    Ok(())
}
```

Either pattern is correct. The second saves one `ExecutionPlan::resolve_field` hash per field when a filter or rename is active.

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

# Engine: TableBuilder

`TableBuilder` (`crates/rypipe-core/src/engine.rs:16`) is the central structure. It implements `ColumnarSink` and is the only production sink that most adapters need.

## Structure

```rust
pub struct TableBuilder {
    pub(crate) columns: Vec<ColumnBuilder>,
    pub(crate) field_index: HashMap<String, usize>,
    pub(crate) column_order: Vec<String>,
    pub(crate) row_count: usize,
    pub(crate) estimated_rows: usize,
    pub(crate) plan: ExecutionPlan,
    pub(crate) row_dirty: Vec<bool>,
}
```

Why this shape:

* `columns: Vec<ColumnBuilder>` holds dense column storage. Indexing `columns[idx]` is a bounds checked array access, not a hash probe. This replaces the earlier `HashMap<String, ColumnBuilder>` that required two hashes per field (one in `ensure_column`, one in `get_mut`). See [Optimizations](./optimizations.md) for the before and after.

* `field_index: HashMap<String, usize>` maps resolved column name to `Vec` index. One hash per field in steady state. `FxHashMap` (rustc_hash) is used for speed on short strings.

* `column_order: Vec<String>` records first appearance order, then reordered by `schema_order` in `sort_columns`. It is independent of `Vec` order, which is insertion order. `schema_insert_index` computes the insertion position for a new column based on the desired output order.

* `row_count: usize` is the number of committed rows. A row is not counted until `finish_row` succeeds (including filter).

* `row_dirty: Vec<bool>` has the same length as `columns`. `row_dirty[i]` is true if column `i` received a value in the current uncommitted row. It lets `finish_row` null fill only missing columns and avoids a per column `while len < target` check for touched columns.

* `estimated_rows: usize` and `plan: ExecutionPlan` are carried from `Pipeline::with_plan` and used for capacity hints and per row decisions.

Constructors (`new`, `with_capacity`, `with_plan`) all initialize the three Vectors and the map as empty.

## Helpers

* `get_column(&self, name: &str) -> Option<&ColumnBuilder>` and `get_column_mut(&mut self, name: &str) -> Option<&mut ColumnBuilder>` are the single lookup path: `field_index.get(name).map(|&i| &self.columns[i])`. Tests and `merge.rs` use these instead of HashMap `get`.

* `take_column(&mut self, name: &str) -> Option<ColumnBuilder>` removes and returns ownership. It does `field_index.remove(name)` to get `idx`, then `columns.swap_remove(idx)` (or `pop` if last). If the removed index was not the last, the element that was at `last` moves to `idx`; the code finds its key in `field_index` (value `== old_last`) and repoints it to `idx`. It also keeps `row_dirty` in sync with `swap_remove` (or `pop`). This is used only by `merge::extend` where `other` is consumed.

## Core row protocol

Adapters call `begin_row`, `put_field` (or `put_field_resolved`), `end_row` in a loop. `TableBuilder` implements these as:

* `begin_row` does nothing. Row boundaries are tracked by `row_count` and `row_dirty`.

* `push_field` resolves the raw name (`plan.field_map` then `plan.drop_fields`) and delegates to `push_field_resolved`. Fast path: if both maps are empty, it uses `name` directly and avoids allocation and hashing in `resolve_field`.

* `push_field_resolved` is the hot path (see [Optimizations](./optimizations.md) for the single lookup version). It calls `ensure_column_idx(resolved)` to get `idx`, marks `row_dirty[idx] = true`, then handles last write wins: if `columns[idx].len() > row_count`, the column already has a value for this row (duplicate field in the same row), so it pops before pushing the new value. Then `push_value` is called on the builder.

* `ensure_column_idx(&mut self, name: &str) -> usize` does one hash lookup. If `field_index.get(name)` exists, it returns immediately. Otherwise it creates a `ColumnBuilder::with_capacity(est, &col_type)` where `est = estimated_rows.max(64)` and `col_type = plan.column_type(name)`, backfills `row_count` nulls (`for _ in 0..row_count { b.push(None) }`), pushes to `columns`, inserts into `field_index`, pushes `false` to `row_dirty`, and inserts into `column_order` at `schema_insert_index(name)`.

* `finish_row` is where the dirty optimization matters (2C-S1). Instead of looping over all columns and doing `while b.len() < target { b.push(None) }`, it does:

```rust
for (i, b) in self.columns.iter_mut().enumerate() {
    if !self.row_dirty[i] {
        b.push(None);
    } else {
        self.row_dirty[i] = false;
    }
}
if let Some(ref filter) = self.plan.filter {
    if !filter.check(&self.columns, &self.field_index, self.row_count, &self.plan) {
        for b in &mut self.columns { b.pop(); }
        return;
    }
}
self.row_count += 1;
```

Only missing columns get a `push(None)`; touched columns are just cleared for the next row. For 10 columns where 8 are present each row, this saves 80% of the null fill pushes and the associated `len` checks. The loop still iterates over `columns.len()` to check the bool, but the bool check is a single byte load versus a `len` load plus branch and push.

Filter is evaluated per row via `FilterPredicate::check` with `(&columns, &field_index, row_index, &plan)`. If it fails, each column is popped (undoing the row) and `row_count` is not advanced. Dirty was already cleared, so the next row starts clean. For `And`/`Or`/`Not` trees, `check` short circuits.

## Other methods

* `reset` clears `columns`, `field_index`, `column_order`, `row_dirty`, and resets `row_count` to zero while keeping `plan` and `estimated_rows`.

* `normalize` truncates any column with `len > row_count` (partial row from a truncated chunk) and clears `row_dirty` to all false. Idempotent.

* `auto_dict_upgrade` iterates `&mut self.columns` and calls `try_upgrade_to_dict(512, max_ratio, max_size)` when `plan.auto_dict` is true. Threshold defaults are 0.05 ratio and 256 entries. It is called from `finish` before sorting.

* `sort_columns` reorders `column_order` by `schema_order` rank. It does not reorder `columns` or `field_index`; those stay insertion ordered and are looked up by name. Only the output order changes.

* `schema_insert_index(&self, name: &str) -> usize` computes where a new column should be inserted into `column_order` to respect `schema_order`. If `schema_order` is empty, it returns `column_order.len()` (append). Otherwise it finds the position of `name` in `schema_order` and returns the position of the first existing column that appears later in that order.

* `finish(&mut self) -> Result<RecordBatch>` (also `ColumnarSink::finish`) does `normalize`, early return with `RecordBatch::new_empty` if `column_order` is empty, `auto_dict_upgrade`, `sort_columns`, then builds `fields` and `arrays` by iterating `column_order` and looking up each builder via `get_column`, calling `arrow_datatype` and `to_arrow_array`. It creates `Arc::new(Schema::new(fields))` and `RecordBatch::try_new`.

## ColumnarSink implementation

```rust
impl ColumnarSink for TableBuilder {
    fn begin_row(&mut self) {}
    fn put_field(&mut self, name: &str, value: Value<'_>) { self.push_field(name, value) }
    fn end_row(&mut self) { self.finish_row() }
    fn wants(&self, name: &str) -> bool { self.resolve(name).is_some() }
    fn resolve<'a>(&'a self, name: &'a str) -> Option<&'a str> { self.plan.resolve_field(name) }
    fn put_field_resolved(&mut self, resolved_name: &str, value: Value<'_>) { self.push_field_resolved(resolved_name, value) }
    fn finish(&mut self) -> Result<RecordBatch> { ... }
}
```

`wants` now delegates to `resolve`, so an adapter that does `if sink.wants(k) { sink.put_field(k, v) }` pays one `resolve_field` hash. The faster pattern is `if let Some(r) = sink.resolve(k) { /* expensive decode */ sink.put_field_resolved(r, v) }` which pays one hash total (see [Decoder](./decoder.md) and [Optimizations](./optimizations.md)).

## Invariants

* `columns.len() == field_index.len() == row_dirty.len()` always.
* `columns.len() == column_order.len()` after each successful `finish_row` or `extend`, but during a row `columns` may be larger than `row_count+1` before `finish_row` completes.
* `row_dirty[i]` is true exactly when `columns[i].len() == row_count + 1` and the column was touched this row; after `finish_row` all entries are false.
* `take_column` keeps the three vectors in sync via swap remove patching.

## Tests

Inside `engine::tests`, `LineParser` plus `LineSplitter` exercise the same `put_field` path used by real adapters. Tests cover `extend` (no duplicates, multi chunk same as single, ragged late debut), last write wins, rename, drop, filter `eq`/`ne`/missing, typed columns, dictionary, and `apply_compare_filter`.

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

# Execution: Pipeline, Parallel, Bounded, Input

`crates/rypipe-core/src/pipeline.rs`, `parallel.rs`, `bounded.rs`, `input.rs` plus `merge.rs` and `plan.rs` decide how bytes become batches.

## Pipeline

```rust
pub struct Pipeline<S, P> { splitter: S, parser: P, plan: ExecutionPlan }
```

`S: Splitter + Clone` and `P: RecordParser + Clone` so the pipeline can be reused across files and modes.

* `new(splitter, parser) -> Self` with `ExecutionPlan::new()`
* `with_plan(plan) -> Self` replaces the plan (builder chain)
* `read_bytes(&self, bytes: &[u8]) -> Result<RecordBatch>` creates `TableBuilder::with_plan(bytes.len() / 512, plan)`, calls `validate` then `parse_chunk` on the whole slice, then `finish`. Single thread, single batch.
* `read_bytes_par(&self, bytes: &[u8], num_chunks: usize) -> Result<Vec<RecordBatch>>` delegates to `ParallelExecutor::parse(bytes, &splitter, parser.clone(), plan, num_chunks)` (no file IO)
* `read_bytes_stream(&self, bytes: &[u8], budget: MemoryBudget) -> Result<Vec<RecordBatch>>` delegates to `BoundedExecutor::new(budget).run_bytes(bytes, &splitter, parser.clone(), plan)`
* `read_path(&self, path: impl AsRef<Path>, use_mmap: bool, prefault: bool) -> Result<RecordBatch>` opens `InputBuffer::open(path, use_mmap, prefault)` and calls `read_bytes(input.as_slice())`
* `read_path_par(&self, path, num_chunks, use_mmap, prefault) -> Result<Vec<RecordBatch>>` opens and calls `ParallelExecutor::parse(input.as_slice(), ...)`
* `read_path_stream(&self, path, budget, prefault) -> Result<Vec<RecordBatch>>` delegates to `BoundedExecutor::run(path, &splitter, parser, plan, prefault)`

All six methods share the same `Splitter` plus `RecordParser` plus `ExecutionPlan`. Tests in `pipeline::tests` use a `LineSplitter` and `LineParser` that split on `\n` and parse `key=value` tokens.

## ParallelExecutor

`crates/rypipe-core/src/parallel.rs:16` `pub struct ParallelExecutor;` with one associated function:

```rust
pub fn parse<P>(bytes: &[u8], splitter: &dyn Splitter, parser: P, plan: ExecutionPlan, num_chunks: usize) -> Result<Vec<RecordBatch>>
where P: RecordParser + Clone + Send + Sync
```

Steps:

1. `splitter.find_split_points(bytes, num_chunks)` then `split_points_to_ranges(&points, bytes.len())` to get `Vec<Range<usize>>`.

2. `into_par_iter` via `rayon` maps each `Range` to `catch_unwind(AssertUnwindSafe(|| { let mut sink = TableBuilder::with_plan(est, plan.clone()); parser.validate(&bytes[range])?; parser.parse_chunk(&bytes[range], &mut sink)?; Ok(sink) }))` where `est = (range.len() / 512).max(64)`. Panics are caught and turned into `Error::Merge("worker panicked during parallel parse: {msg}")` by downcasting `payload` to `&str` or `String`.

3. `collect::<Result<Vec<TableBuilder>>>()` joins. If `engines` is empty, return `Ok(vec![])`.

4. Fast path: `if !plan.auto_dict && schemas_consistent(&engines) { return engines_to_record_batches(engines, &plan) }`

   `schemas_consistent` builds `base: HashMap<&str, &str>` from `first.field_index` plus `first.columns[idx].variant_key()` and checks every other engine's `field_index` entries have the same `variant_key`. Missing columns are fine (null filled later). This allows `int64` plus `float64` to be considered inconsistent here (so merge path will promote), but `string` plus `dictionary` is also inconsistent and will promote in the fast path via `unify_variants` (not here). Actually `schemas_consistent` requires exact key equality, so `int64` vs `float64` fails and falls to merge path which also promotes; `string` vs `dictionary` also fails but fast path `engines_to_record_batches` handles promotion as well, so the fast path is still taken when `auto_dict` is false? Wait, `schemas_consistent` returning true requires exact match, so mixed `string`/`dictionary` would be false and go to merge path even though `engines_to_record_batches` could handle it. Current code does: fast path only if `!auto_dict && schemas_consistent`. That means `string` plus `dictionary` with `auto_dict` false but different variants will go to merge path (single batch) instead of fast path (multiple batches with unified schema). This is intentional to keep `engines_to_record_batches` as the unified schema path; the merge path also handles it but with single batch. The doc says fast path emits one batch per chunk with unified schema; merge path returns single merged batch. Both handle promotion, but fast path keeps chunked batches.

5. Merge path: `let mut merged = TableBuilder::with_plan(engines.len().max(64) * 512, plan.clone()); for engine in engines { merged.extend(engine)?; } let batch = merged.finish()?; if let Some(filter) = plan.filter { return Ok(vec![apply_compare_filter(batch, filter)?]) }` Note `apply_compare_filter` is only applied here for the merged single batch; fast path applies it per batch inside `engines_to_record_batches`.

All row filters (`Equal`, `NotEqual`, `Compare`, and `And`/`Or`/`Not` trees) are evaluated per row during `finish_row` in both paths, so they never force the merge path.

## BoundedExecutor

`crates/rypipe-core/src/bounded.rs:14` `MemoryBudget` is `bytes: usize` with `new` and `bytes()`.

`BoundedExecutor { budget: MemoryBudget }` has:

* `plan_chunks(&self, bytes: &[u8], splitter: &dyn Splitter) -> (Vec<Range<usize>>, usize, usize)` estimates `bytes_per_row = splitter.estimate_bytes_per_row(bytes).max(1)`, `total_rows_est = bytes.len() / bytes_per_row`, `rows_per_batch = (budget.bytes() / bytes_per_row).max(1).min(total_rows_est.max(1))`, `num_batches = (total_rows_est / rows_per_batch).max(1)`, `split_points = splitter.find_split_points(bytes, num_batches.min(MAX_SPLIT_CHUNKS))` where `MAX_SPLIT_CHUNKS = 256`, then `split_points_to_ranges`.

* `run_bytes<P>(&self, bytes: &[u8], splitter: &dyn Splitter, parser: P, plan: ExecutionPlan) -> Result<Vec<RecordBatch>>` where `P: RecordParser + Clone + Send + Sync`. For empty bytes returns `Ok(vec![])`. Otherwise it gets `(chunks, rows_per_batch, bytes_per_row)`, creates `batch_engine = TableBuilder::with_plan(bytes_per_row.max(64), plan)`, then for each `chunk` slices `&bytes[chunk.start..chunk.end]`, creates a per chunk `chunk_engine`, calls `validate` and `parse_chunk`, extends `batch_engine` via `extend`, tracks `rows_in_batch`, flushes when `rows_in_batch >= rows_per_batch` via `batch_engine.finish()` plus `reset`. At the end flushes remainder and calls `apply_plan_filter` (which applies `apply_compare_filter` only for pure `Compare` and `And` trees; other trees are no ops because per row is authoritative).

* `run<P>(&self, path: &Path, splitter: &dyn Splitter, parser: P, plan: ExecutionPlan, prefault: bool) -> Result<Vec<RecordBatch>>` opens `InputBuffer::open(path, use_mmap = cfg(feature="mmap"), prefault)`. If the buffer is `Mmap`, it calls `run_mapped` which does `plan_chunks` on the mapped slice, drops the mapping, then reopens the file with `File::open` and for each `chunk` does `seek` plus `read_exact` into a fresh `Vec<u8>`, parses, and accumulates as above. This keeps RSS low for large files: the mapping is released before the parse loop, and only one chunk buffer is live at a time. If the buffer is `Owned` (including transparently decompressed), it delegates to `run_bytes(input.as_slice(), ...)`.

* `run_mapped` is `#[cfg(feature="mmap")]` and takes `input: InputBuffer` by value (so the mapping is dropped after `plan_chunks`). The file is reopened; chunk reads use `SeekFrom::Start(chunk.start)` plus `read_exact`.

`MAX_SPLIT_CHUNKS = 256` is the internal safeguard: never request more than 256 split points even if `budget` would imply more batches; pathological `bytes_per_row` cannot explode per chunk overhead. Batches may still exceed budget when the required count exceeds the cap (documented).

## InputBuffer

`crates/rypipe-core/src/input.rs:36` `enum InputBuffer { Mmap(MmapHandle), Owned(Vec<u8>) }` where `MmapHandle` wraps `memmap2::Mmap`.

* `MmapHandle::new(file, prefault)` maps the file and on Unix does `mmap.advise(WillNeed)` if `prefault` else `Sequential`.

* `detect_compression(path) -> Option<Compression>` reads the first 4 bytes and matches magic: `gzip` `1f 8b` (2 bytes), `zstd` `28 b5 2f fd`, `lz4` frame `04 22 4d 18`. Each arm is `#[cfg(feature = "gzip"/"zstd"/"lz4")]` so detection only fires when the feature is enabled. No extension check, only magic.

* `decompress(path, codec) -> Result<Vec<u8>>` opens the file again and wraps it in `flate2::read::GzDecoder`, `zstd::stream::read::Decoder`, or `lz4_flex::frame::FrameDecoder` depending on codec and feature, then `read_to_end`.

* `open(path: &Path, use_mmap: bool, prefault: bool) -> Result<Self>` first calls `detect_compression`; if `Some`, returns `Owned(decompress(...)?)` (so all execution modes operate on decompressed bytes). Otherwise, if `#[cfg(feature="mmap")]` and `use_mmap`, returns `Mmap`; else reads via `fs::read` into `Owned`.

* Cargo features: `gzip = ["dep:flate2"]`, `zstd = ["dep:zstd"]`, `lz4 = ["dep:lz4_flex"]`, `compress-all = ["gzip","zstd","lz4"]`, `mmap = ["dep:memmap2"]`. The `zstd` and `lz4` decoders are pure Rust when possible (`flate2` with `rust_backend`).

## Merge

`crates/rypipe-core/src/merge.rs:14` `impl TableBuilder { extend, }` plus `engines_to_record_batches`.

* `extend(&mut self, mut other: TableBuilder) -> Result<()>` merges `other` into `self`. Steps: (1) for each name in `other.column_order.clone()` where `!self.field_index.contains_key(name)`, create a builder `with_capacity(est, &col_type)` where `est = self_rows + other.estimated_rows.max(64)`, backfill `self_rows` nulls, push to `columns`, insert to `field_index`, push false to `row_dirty`, insert into `column_order` at `schema_insert_index`. (2) snapshot `order_snapshot = self.column_order.clone()`, then for each `name` in `order_snapshot` get `self_idx` via `field_index`, take `self_b = &mut columns[self_idx]`, try `other.take_column(name)`; if `Some`, check `variant_key` equality, call `unify_variants` if different (string plus dictionary to dictionary, int64 plus float64 to float64 else `Error::Merge` with column name and hint to provide `field_types`), then `promote_to_variant` on both, then `extend_owned`; else null pad `other_rows` times. Finally `row_count = self_rows + other_rows`.

* `engines_to_record_batches(mut engines: Vec<TableBuilder>, plan: &ExecutionPlan) -> Result<Vec<RecordBatch>>` exports per chunk builders without serial merge. It normalizes and retains `row_count > 0`, builds `order` plus `targets: HashMap<String, &'static str>` via `get_column` and `unify_variants` folding, promotes each builder's columns to the unified variant, builds `types: HashMap<String, DataType>` from first sighting `arrow_datatype`, creates `Schema`, then `par_iter` over engines to build `arrays` per `order` (via `get_column` or `null_array` for missing), `RecordBatch::try_new`, collects via `rayon`, then applies `apply_compare_filter` per batch if `plan.filter` is Some.

## Arrow export

`crates/rypipe-core/src/arrow_export.rs` `null_array`, `apply_compare_filter`, `compare_columns`, `is_numeric`.

* `apply_compare_filter(batch, predicate)` is only for pure `Compare` and `And` trees (checked via `is_pure_compare_tree`). Other trees return `Ok(batch)` unchanged because per row is authoritative. For pure trees, it builds a mask via `compare_mask` (recursing `And` with `and` kernel) and `compare_columns` (casts both to `Float64` if numeric else `Utf8`, then Arrow `gt`, `lt`, `gt_eq`, `lt_eq`, `eq`, `neq`), then `filter_record_batch`.

See also [Engine](./engine.md) for `TableBuilder::finish` and Arrow `to_arrow_array` details per `ColumnBuilder` variant.

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

# Architecture Overview

`rypipe` is built around one idea: **separate the parts of parsing that depend on a file format from the parts that do not.** The format specific side answers where a row ends and how to extract fields; the format agnostic side answers how to store rows as typed columns, how to project them, and how to get them out as Arrow.

This directory documents the architecture in depth. If you are new, read this page first; then dive into the subpages for the component you care about.

* [Engine (TableBuilder)](./engine.md): row handling, last write wins, dirty tracking, column dispatch
* [Columnar storage](./columnar.md): `StrColumn`, `ColumnBuilder` variants, dictionary, auto dict, promotions
* [Execution plan](./plan.md): `ExecutionPlan`, `FieldType`, `FilterPredicate` trees
* [Execution (Pipeline, Parallel, Bounded)](./execution.md): `Pipeline`, `ParallelExecutor`, `BoundedExecutor`, `InputBuffer`
* [Decoder API](./decoder.md): `Splitter`, `RecordParser`, `ColumnarSink` with `resolve` and `put_field_resolved`
* [Data flow](./data-flow.md): diagrams for single, parallel, and bounded modes
* [Optimizations](./optimizations.md): every optimization, why it matters, and what it replaces
* [Storage and export](./storage.md): Arrow export, null handling, and compare filter reapplication

## Design philosophy

The engine was extracted from a single format (Crystal Reports XML) and generalized so that the same hot path serves every adapter. The split is deliberate and strict:

**Adapter bound (format specific, lives in adapter crates, you implement it):** where does one row end, how do I extract field names and values, what is the encoding or entity rule. This code knows the file format and nothing about column storage.

**Core (format agnostic, lives in `rypipe` crates, reused by every adapter):** how do I store rows as typed columns, how do I rename, drop, cast, filter, and reorder, how do I export to Arrow, how do I parallelize while staying inside a memory budget. This code knows the columnar engine and nothing about any file format.

Adapters live in separate packages. `rypipe` ships zero parsers. The boundary is a trait, not a stringly convention.

## High level flow

Legend: `(ADAPTER BOUND)` = you write it in the adapter crate; `(CORE)` = lives in `rypipe` and is reused unchanged.

```
input bytes  [ADAPTER or CORE decides the source; InputBuffer below]
    |
    v
+---------------------+  (ADAPTER BOUND) format specific
|   Splitter          |  Trait you implement
|                     |  find_split_points, estimate_bytes_per_row
+----------+----------+
           | Vec<Range<usize>>  (CORE) helper split_points_to_ranges
           v
+---------------------+  (ADAPTER BOUND) format specific
| RecordParser        |  Trait you implement
|                     |  validate, parse_chunk -> begin_row, put_field, end_row
+----------+----------+
           | Value events  (CORE) enum Value<'a>
           | Str, Int64, Float64, Bool, Date32, Timestamp, Null
           v
+---------------------+  (CORE) format agnostic
|  TableBuilder       |  Struct in rypipe-core/src/engine.rs
|  (ColumnarSink)     |  columns: Vec<ColumnBuilder> (CORE)
|                     |  field_index: HashMap<String,usize> (CORE)
|                     |  column_order: Vec<String> (CORE)
|                     |  row_dirty: Vec<bool> (CORE)
|                     |  plan: ExecutionPlan (CORE)
+----------+----------+
           | RecordBatch  (CORE) Arrow
           v
+---------------------+  (CORE) format agnostic
|  Arrow export       |  rypipe-core/src/arrow_export.rs plus engine.rs:finish
|                     |  to_arrow_array, apply_compare_filter, null_array
+---------------------+
           | pyarrow.Table / pandas / Polars  [PYTHON CORE] via rypipe-python C Data Interface
           v
      downstream (join, aggregate, plot) [PYTHON / ADAPTER CONSUMER]
```

Ownership rule: `Splitter` plus `RecordParser` never see `ExecutionPlan` or `TableBuilder` internals. They emit plain names and `Value` events; the engine resolves names (`ExecutionPlan::resolve_field`) and chooses storage (`column_type`). This keeps adapters tiny and keeps all pushdown, typing, and Arrow logic in one place.

## Crate overview

All format specific code is **ADAPTER BOUND** (separate crates, not in this repo). All rows below are **CORE** (format agnostic, in `rypipe`).

### `rypipe-core` (pure Rust, no pyo3, no quick-xml) (CORE)

| Module | File | Side | Responsibility |
|--------|------|------|----------------|
| `value` | `value.rs` | CORE | `Value<'a>` enum: `Str(&str)`, `Int64`, `Float64`, `Bool`, `Date32(i32)`, `Timestamp(i64)`, `Null` |
| `plan` | `plan.rs` | CORE | `ExecutionPlan`, `FieldType` (String, Int64, Float64, Boolean, Dictionary, Date32, Timestamp(unit)), `FilterPredicate` (Equal, NotEqual, Compare, And, Or, Not), `CompareOp` |
| `columnar` | `columnar.rs` | CORE | `StrColumn` (arena plus offsets plus validity), `ColumnBuilder` (7 variants), dictionary `value -> i32` index, `try_upgrade_to_dict`, `unify_variants`, `promote_to_variant` |
| `engine` | `engine.rs` | CORE | `TableBuilder` with `Vec<ColumnBuilder>`, `field_index`, `column_order`, `row_count`, `row_dirty`, `estimated_rows`, `plan`; `ensure_column_idx`, `push_field_resolved`, `push_field`, `finish_row`, `normalize`, `auto_dict_upgrade`, `sort_columns`, `finish` |
| `decoder` | `decoder.rs` | CORE (traits) / ADAPTER BOUND (impls) | Traits `Splitter`, `RecordParser`, `ColumnarSink` (with `resolve` and `put_field_resolved`) plus `split_points_to_ranges`; adapters implement the first two, core implements the third |
| `pipeline` | `pipeline.rs` | CORE | `Pipeline<S,P>` wiring `Splitter` plus `RecordParser` to `TableBuilder` with `read_bytes`, `read_bytes_par`, `read_bytes_stream`, `read_path`, `read_path_par`, `read_path_stream` |
| `parallel` | `parallel.rs` | CORE | `ParallelExecutor::parse` with rayon, fast path vs merge path, `schemas_consistent` |
| `bounded` | `bounded.rs` | CORE | `BoundedExecutor` plus `MemoryBudget`, `run`, `run_bytes`, `run_mapped`, `plan_chunks`, `MAX_SPLIT_CHUNKS = 256` |
| `input` | `input.rs` | CORE | `InputBuffer` (`Mmap` or `Owned`), `MmapHandle`, magic byte detection for `gzip` (`1f 8b`), `zstd` (`28 b5 2f fd`), `lz4` frame (`04 22 4d 18`), transparent decompression |
| `merge` | `merge.rs` | CORE | `TableBuilder::extend`, `engines_to_record_batches` with variant unification and promotion |
| `arrow_export` | `arrow_export.rs` | CORE | `null_array`, `apply_compare_filter` (pure Compare and And only; other trees are no ops), `compare_columns`, `is_numeric` |
| `error` | `error.rs` | CORE | `Error` enum (`Utf8`, `Plan(String)`, `Merge(String)`, `Io`, `Arrow`) and `Result` |
| `lib` | `lib.rs` | CORE | Reexports |

**Not in `rypipe-core`:** `XmlSplitter`, `CsvSplitter`, `JsonSplitter`, `CrystalXmlDecoder` etc. are **ADAPTER BOUND** and live in separate packages (see `docs/crxml-adapter.md` and `docs/writing-adapters.md`). They depend on `rypipe-core` but `rypipe-core` never imports them.

### `rypipe-python` (PyO3 bindings) (CORE)

| Module | File | Side | Responsibility |
|--------|------|------|----------------|
| `lib.rs` | `crates/rypipe-python/src/lib.rs` | CORE | `_rypipe` extension, exception types (`ParseError`, `XmlError`, `PlanError`, `MergeError`), `py_err_from_rypipe` |
| `plan_kwargs.rs` | `plan_kwargs.rs` | CORE | `execution_plan_from_kwargs` converting Python kwargs to `ExecutionPlan` with nested filter trees (`and`, `or`, `not`) |
| `export.rs` | `export.rs` | CORE | `record_batch_to_pyarrow`, `record_batches_to_pyarrow_batches`, `record_batches_to_pyarrow_table` |

### Python package `rypipe` (CORE)

| Module | File | Side | Responsibility |
|--------|------|------|----------------|
| `source` | `rypipe/source.py` | CORE | `Source` abstract base, `Adapter` convenience base (adapters subclass it; this file itself is core) |
| `pipeline` | `rypipe/pipeline.py` | CORE | `Pipeline` stage chaining (`|`), `plan_split` fusion, `to_arrow` short circuit |
| `fusion` | `rypipe/fusion.py` | CORE | `plan_split` (multi filter `and` combine), `_try_columnar_fusion`, `fused_iter` |
| `batchpipe` | `rypipe/batchpipe.py` | CORE | `Batch`, `Operator`, `ArrowSource`, `FusedTransforms`, `LambdaOp`, `build_chain`, `iter_dicts`, `collect_table` |
| `stages` | `rypipe/stages/*.py` | CORE | `RenameFields`, `DropFields`, `CastTypes`, `FilterRows`, `FilterRowsAny`, `FilterRowsAll`, `FilterRowsNot` |
| `sinks` | `rypipe/sinks.py` | CORE | `collect`, `to_arrow`, `to_dataframe`, `to_polars`, `to_parquet`, `to_csv` |

**ADAPTER BOUND** on the Python side: `my_adapter.MySource`, `my_adapter.MyAdapter`, `CrystalXmlSource` etc. They subclass `Source`/`Adapter` and are not in `rypipe`.

## How the boundary is enforced

`RecordParser` never sees `ExecutionPlan`. It emits plain field names and `Value` events. `TableBuilder` (through `ColumnarSink::put_field` or `put_field_resolved`) resolves those names via `ExecutionPlan::resolve_field` (rename then drop) and chooses storage via `ExecutionPlan::column_type`. This keeps adapters tiny and keeps all pushdown logic in one place.

`ColumnarSink::wants` and the newer pair `resolve` plus `put_field_resolved` let adapters skip dropped fields with one hash lookup instead of two. See [Decoder API](./decoder.md) and [Optimizations](./optimizations.md).

## State and ownership

`TableBuilder` owns three parallel structures: `columns: Vec<ColumnBuilder>` (dense storage), `field_index: HashMap<String, usize>` (name to Vec index), and `column_order: Vec<String>` (output order). A fourth vector `row_dirty: Vec<bool>` tracks which columns were touched in the current row (see [Engine](./engine.md) and [Optimizations](./optimizations.md)). All are cleared together in `reset` and kept in sync in `take_column` (swap_remove with index patching) and `extend`.

`Pipeline<S,P>` is `Clone` where `S: Splitter + Clone` and `P: RecordParser + Clone`, so the same pipeline can be reused across files and execution modes.

## Next steps

Start with [Engine](./engine.md) if you want the hot path, [Columnar](./columnar.md) if you want storage, or [Execution](./execution.md) if you want scheduling. [Optimizations](./optimizations.md) explains every change from the original `HashMap`-based design and why it matters for all adapters, not just one format.

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

# Optimizations

This page lists every optimization that makes `rypipe` fast for all adapters, not just one format. Each entry gives the file and line, what changed, why it matters, and what it replaces.

## 1. Vec plus map column storage (engine.rs:16, plan.rs:280, merge.rs:30, parallel.rs:101)

**Before:** `columns: HashMap<String, ColumnBuilder>` with `HashMap::get_mut` per field. `push_field` did `ensure_column` (`contains_key` plus `insert`) then `get_mut` (second hash) : 2 hashes per field in steady state.

**After:** `columns: Vec<ColumnBuilder>` plus `field_index: HashMap<String, usize>` and `column_order: Vec<String>`. `ensure_column_idx` does `field_index.get(name)` (one hash) and returns `idx`; `push_field_resolved` then does `&mut columns[idx]` (array index). Hot path is one hash plus one indexed load. `take_column` keeps the three vectors in sync via `swap_remove` with index patching.

**Why generic:** every `RecordParser` calls `put_field` for every field of every row. The `Pipeline` hot loop is `validate` plus `parse_chunk` plus `push_value` plus this dispatch. Fewer hashes helps TSV at 5 fields per row as much as XML at 20 fields.

## 2. Single lookup push (engine.rs:202)

**Before:** `ensure_column` plus `get_mut` : double hash as above, plus `schema_insert_index` inside `ensure_column` scanned `column_order` even when the column already existed.

**After:** `ensure_column_idx` is the single lookup. `push_field_resolved` marks `row_dirty[idx] = true` and handles last write wins with one `len > row_count` check. `schema_insert_index` is only called when the column is new. The old `ensure_column` (void) is removed; `push_field` delegates to `push_field_resolved` after one `resolve_field`.

**Why not `entry` API:** `HashMap::entry` would also be one hash, but `Vacant` insertion still needs `schema_insert_index` which borrows `self` mutably while `entry` holds a borrow. The `get` plus `insert` split avoids the borrow checker fight and keeps the fast path as a pure `get` without allocating the key.

## 3. Dirty bitmask for finish_row (engine.rs:274)

**Before:** `for b in &mut columns { while b.len() < target { b.push(None) } }` iterated all columns and pushed `None` for missing ones, checking `len < target` for every column.

**After:** `row_dirty: Vec<bool>` has the same length as `columns`. `push_field_resolved` sets `row_dirty[idx] = true` on first touch of the row (idempotent for duplicate fields). `finish_row` does `for (i,b) in columns.iter_mut().enumerate() { if !row_dirty[i] { b.push(None) } else { row_dirty[i]=false } }`. Only missing columns get a push; touched columns are just cleared. `reset` and `normalize` clear the mask, `take_column` keeps it in sync via `swap_remove`.

**Impact:** for 10 columns where 8 are present, this saves 80% of the `push(None)` calls and the associated `len` loads. The loop still iterates over `columns.len()` to check the bool, but a byte load plus branch is cheaper than a `Vec` push.

## 4. Resolve plus put_field_resolved (decoder.rs:46, engine.rs:302)

**Before:** adapters did `if sink.wants(k) { /* decode */ sink.put_field(k, v) }` which called `ExecutionPlan::resolve_field` twice: once in `wants` (hash `field_map` then `drop_fields`) and once in `put_field`. When `field_map` or `drop_fields` is non empty, this is two hashes per field. `push_field` also did `owned = n.to_owned()` to hold the resolved name.

**After:** `ColumnarSink` has `resolve<'a>(&'a self, name: &'a str) -> Option<&'a str>` and `put_field_resolved(&mut self, resolved, value)`. `TableBuilder` implements `resolve` as `self.plan.resolve_field(name)` (borrows from `field_map` when renamed) and `put_field_resolved` as `push_field_resolved` (no re hash). Adapters can do `if let Some(r) = sink.resolve(k) { /* expensive decode */ sink.put_field_resolved(r, v) }` with one hash total. `wants` now delegates to `resolve` so old code still gets the single hash path.

**Why not `wants_resolved`:** the pair `resolve` plus `put_field_resolved` is the minimal extension that keeps the trait backward compatible (defaults delegate to `put_field`, which re resolves). Adapters choose the fast path only when extraction is expensive; stringly adapters like `LineParser` keep the simple `wants` plus `put_field` path.

## 5. Filter check with Vec plus index (plan.rs:280)

**Before:** `FilterPredicate::check(&self, columns: &HashMap<String, ColumnBuilder>, ...)` did `columns.get(resolve(field, plan))` which hashed the resolved name per filter field per row.

**After:** `check(&self, columns: &[ColumnBuilder], field_index: &HashMap<String, usize>, ...)` does `get_column(columns, field_index, resolve(field, plan))` which is `field_index.get(name).map(|&i| &columns[i])` : one hash per leaf, shared with the column dispatch hash, and no clone of `String` keys. `get_value` and `Compare` arms use the same helper.

**Why generic:** every row that has a filter pays this per row. `Equal`/`NotEqual` plus `Compare` plus `And`/`Or`/`Not` trees all use it. Vectorizing `Equal`/`NotEqual` after `finish` would be possible (see `arrow_export`), but per row is authoritative for short circuiting and missing field semantics, so the Vec path is the right intermediate.

## 6. Unified schema with promotion (columnar.rs:53, merge.rs:30)

**Before:** mixed `int64` plus `float64` or `string` plus `dictionary` across chunks was an error or silent first sighting.

**After:** `unify_variants` plus `promote_to_variant` reconcile `int64` plus `float64` to `float64` and `string` plus `dictionary` to `dictionary` before `extend_owned` or `engines_to_record_batches`. Irreconcilable types return `Error::Merge` naming the column and hinting to provide `field_types`. This lets `ParallelExecutor` fast path keep chunked batches with one unified schema instead of forcing the merge path.

## 7. InputBuffer magic and decompression (input.rs:33, Cargo.toml:15)

**Before:** only `Mmap` or `fs::read`, no compression.

**After:** `detect_compression` reads 4 bytes, matches `1f 8b` (gzip), `28 b5 2f fd` (zstd), `04 22 4d 18` (lz4 frame) when the corresponding Cargo feature is enabled (`gzip`, `zstd`, `lz4`, `compress-all`). `open` then returns `Owned(decompress(...))` via `flate2`, `zstd`, or `lz4_flex`. Bytes APIs slice directly; `BoundedExecutor::run` uses `run_bytes` for `Owned` and `run_mapped` (seek plus read) for `Mmap`.

## 8. Bounded executor chunking (bounded.rs:30, pipeline.rs:65)

**Before:** `Pipeline` had only `read_bytes` (single) and `read_path` variants; bounded mode required a file path and did `File::open` plus `seek`.

**After:** `Pipeline::read_bytes_par` and `read_bytes_stream` plus `BoundedExecutor::run_bytes` slice directly from `&[u8]` with no file IO. `plan_chunks` computes `bytes_per_row`, `total_rows_est`, `rows_per_batch`, `num_batches`, caps at `MAX_SPLIT_CHUNKS = 256`, then `split_points_to_ranges`. `run_bytes` is used for decompressed buffers and for adapters that hold `BytesIO` data.

## 9. Arrow export and filter reapplication (arrow_export.rs:29, engine.rs:289)

* `StrColumn` arena plus offsets plus validity is block copied to `StringArray` (two buffers).
* Numeric builders are dense `Vec<Option<T>>` with `collect`.
* `apply_compare_filter` only reapplies pure `Compare` and `And` of `Compare` via Arrow compute; other trees are no ops because per row is authoritative. This avoids double null semantics and keeps the fast path correct.

## 10. Small but cumulative

* `estimated_rows` capacity hint (`bytes.len() / 512`, min 64) in `TableBuilder::with_plan` reduces reallocations.
* `lexical::parse` for numbers instead of `str::parse` (faster).
* `simdutf8::basic::from_utf8` for validate.
* `FxHashMap` and `FxHashSet` for field names (short strings).
* `with_capacity` for `StrColumn` arena (`cap * 16` bytes heuristic).
* `ColumnarSink` defaults keep the trait object safe (`&mut dyn ColumnarSink` in `parse_chunk`).

Each of these is 0.5 to 2%, but together they move single thread from ~600 MB/s on the original `HashMap` plus double lookup plus `while len < target` baseline to the current 700 MB/s, and filtered cases by 8 to 12% via `resolve` plus `put_field_resolved`. None of them require `unsafe`; all are behind safe API changes. When safe leaves more than a few percent on the table, the remaining `unsafe` candidates are gated behind `#[cfg(feature="unsafe-fast")]` and require `miri` plus `cargo test --release` proof, per the plan in the repo root.

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

# Execution plan

`crates/rypipe-core/src/plan.rs` is the pushdown target. Everything an adapter can fuse is expressed here.

## ExecutionPlan

```rust
pub struct ExecutionPlan {
    pub field_map: HashMap<String, String>,
    pub drop_fields: HashSet<String>,
    pub field_types: HashMap<String, FieldType>,
    pub dictionary_columns: HashSet<String>,
    pub filter: Option<FilterPredicate>,
    pub schema_order: Vec<String>,
    pub auto_dict: bool,
    pub dict_threshold: Option<f64>,
    pub dict_max_size: Option<usize>,
}
```

Default is a no op. Builder methods (`rename`, `drop`, `drop_many`, `type_as`, `dictionary`, `filter_eq`, `filter_ne`, `filter_compare`, `schema_order`, `with_auto_dict`, `with_dict_threshold`, `with_dict_max_size`) each insert or set the field and return `Self` for chaining. Fields are public for direct mutation when needed.

Resolution order (also documented in `docs/architecture/index.md`):

1. `field_map` renames raw field (`raw -> output`)
2. `drop_fields` checks the resolved name (`None` means drop)
3. `field_types` or `dictionary_columns` chooses storage (`column_type`)
4. `filter` rejects rows in `finish_row`
5. `schema_order` reorders `column_order` in `finish`

## FieldType

```rust
pub enum FieldType { String, Int64, Float64, Boolean, Dictionary, Date32, Timestamp(TimeUnit) }
```

`from_str` parses: `string`, `int64`, `float64`, `bool` or `boolean`, `dictionary`, `date32`, `timestamp` (defaults to Microsecond), `timestamp[s]`, `timestamp[ms]`, `timestamp[us]` or `timestamp[µs]`, `timestamp[ns]`.

`column_type(&self, name: &str) -> FieldType` checks `field_types` first, then `dictionary_columns`, else `String`. This is called once per distinct column at first use (in `ensure_column_idx`), not per row.

## FilterPredicate

```rust
pub enum FilterPredicate {
    NotEqual { field: String, value: String },
    Equal { field: String, value: String },
    Compare { field_a: String, op: CompareOp, field_b: String },
    And(Box<FilterPredicate>, Box<FilterPredicate>),
    Or(Box<FilterPredicate>, Box<FilterPredicate>),
    Not(Box<FilterPredicate>),
}
```

`Compare` is column to column with numeric promotion; `Equal`/`NotEqual` are constant string comparisons. `And`/`Or`/`Not` compose trees arbitrarily. Helpers `FilterPredicate::all`, `any`, `not` build boxed trees.

`CompareOp` (`Gt`, `Lt`, `Ge`, `Le`, `Eq`, `Ne`) parses from `>`, `gt`, `<`, `lt`, `>=`, `ge`, `<=`, `le`, `==`, `eq`, `!=`, `ne`.

## Resolve

```rust
pub fn resolve_field<'a>(&'a self, raw: &'a str) -> Option<&'a str> {
    let resolved = self.field_map.get(raw).map_or(raw, |s| s.as_str());
    if self.drop_fields.contains(resolved) { return None; }
    Some(resolved)
}
```

Rename first, then drop on the resolved name. Fast path in `TableBuilder::push_field` skips this entirely when both maps are empty (`field_map.is_empty() && drop_fields.is_empty()`), so the hot keep-all case pays zero hash.

When maps are not empty, each field still pays one or two hashes. The optimization in `decoder::ColumnarSink` (`resolve` plus `put_field_resolved`) lets adapters pay it once instead of twice (`wants` plus `put_field`). See [Decoder](./decoder.md) and [Optimizations](./optimizations.md).

## Check

```rust
pub(crate) fn check(&self, columns: &[ColumnBuilder], field_index: &HashMap<String, usize>, row_index: usize, plan: &ExecutionPlan) -> bool
```

This is the per row filter used in `TableBuilder::finish_row`. It takes the `Vec<ColumnBuilder>` plus `field_index` (single hash per field) instead of a `HashMap` view (which would double hash). Steps:

* `Equal`/`NotEqual` call `get_value` which does `get_column(columns, field_index, resolve(field, plan)).and_then(|b| b.get_filter_value(row_index))` and compares `Option<String>` to the constant.

* `Compare` fetches `get_column` for each side, then `get_typed_value(row_index)` to get `TypedValue`, then `compare_typed(&a, op, &b)`. `compare_typed` promotes `Int64` vs `Float64` to `f64` via `partial_cmp`; other same type pairs use `Ord`; mixed non numeric or different temporal units return false. Null operands fail the row (caller checks `Some`).

* `And` short circuits on first failure, `Or` on first success, `Not` negates. A missing field fails the inner leaf, which `Not` can flip to keep. There is a `#[cfg(test)]` shim `check_map` that preserves old HashMap semantics for tests that still hold a map view.

`check` is called with `row_index == self.row_count` (the uncommitted row) before `row_count` is advanced. If it returns false, `finish_row` pops every column (undo) and returns without counting the row.

## Python surface

`crates/rypipe-python/src/plan_kwargs.rs` converts Python kwargs to `ExecutionPlan`. It accepts `filter` as a dict that may be a leaf (`field`, `op`, `value` or `field_a`, `op`, `field_b`) or a tree (`and: [spec, ...]`, `or: [spec, ...]`, `not: spec`) and recurses via `parse_filter_spec`, `combine_list`, `parse_leaf_spec`. Valid ops and types are validated there and surface as `PlanError`.

On the Python side, `FilterRows` plus `FilterRowsAny`, `FilterRowsAll`, `FilterRowsNot` build those dicts via `_plan_kwargs`, and `fusion::plan_split` merges multiple `filter` specs into `{"and": [...]}` so chaining `FilterRows` stages does not drop the earlier filter.

## Schema order

`schema_order: Vec<String>` is the desired output column order. Columns not listed keep first appearance order after the listed ones. `TableBuilder::sort_columns` and `schema_insert_index` implement this. The order does not affect `field_index` or `columns` Vec order, only `column_order`.

## Auto dict

`auto_dict: bool` plus `dict_threshold` (default 0.05) and `dict_max_size` (default 256) feed `ColumnBuilder::try_upgrade_to_dict`. See [Columnar](./columnar.md) for the heuristic and [Execution](./execution.md) for fast vs merge path interaction.

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

# Storage and export

This page covers `crates/rypipe-core/src/columnar.rs:34` `StrColumn`, `ColumnBuilder`, dictionary, and `crates/rypipe-core/src/arrow_export.rs` plus the `finish` path in `engine.rs:289`.

## Arrow types produced

| `ColumnBuilder` variant | Arrow `DataType` | Array type |
|-------------------------|-----------------|------------|
| `String(StrColumn)` | `Utf8` | `StringArray` (OffsetBuffer plus Buffer plus NullBuffer) |
| `Int64(Vec<Option<i64>>)` | `Int64` | `Int64Array` |
| `Float64(Vec<Option<f64>>)` | `Float64` | `Float64Array` |
| `Boolean(Vec<Option<bool>>)` | `Boolean` | `BooleanArray` |
| `Date32(Vec<Option<i32>>)` | `Date32` | `Date32Array` |
| `Timestamp(TimeUnit, Vec<Option<i64>>)` | `Timestamp(unit, None)` | `PrimitiveArray<TimestampSecondType>` etc. |
| `Dictionary { codes, dict, index }` | `Dictionary(Int32, Utf8)` | `DictionaryArray<Int32Type>` with `Int32Array` keys and `StringArray` values |

`arrow_datatype(&self) -> DataType` and `to_arrow_array(&self) -> Result<ArrayRef>` implement the mapping. `Timestamp` branches on `TimeUnit` to the correct `PrimitiveArray` type.

## Null handling

* `StrColumn` has `validity: Vec<bool>`. `to_arrow` builds `NullBuffer` only if some validity is false; otherwise `None` (all valid). Offsets still advance by 0 for null entries so `offsets[i]==offsets[i+1]`.

* Numeric and boolean builders are `Vec<Option<T>>`. `collect::<Int64Array>()` etc. preserves nulls.

* Missing columns in `engines_to_record_batches` become `null_array(&types[name], e.row_count)` (a `NullArray` of the unified type).

* `Value::Null` and unparseable strings both become `None`.

## String arena

`StrColumn::push(Option<&str>)` appends bytes to `data` and `data.len()` to `offsets`. `pop` truncates `data` to `offsets.last()`. `append` merges another column with base shifted offsets. `get` slices `data[offsets[i]..offsets[i+1]]` and does `from_utf8` (safe because input was validated via `simdutf8`).

Capacity: `with_capacity(cap)` reserves `cap+1` offsets and `cap*16` data bytes. `Pipeline` passes `cap = bytes.len() / 512` (min 64) or `estimated_rows`.

## Numeric and temporal parsing

`push` and `push_str` for typed builders use:

* `lexical::parse::<i64,_>` and `<f64,_>` for `Int64`/`Float64`
* `s.parse::<bool>()` for `Boolean`
* `parse_date32` (`chrono::NaiveDate::parse_from_str("%Y-%m-%d")` minus epoch) for `Date32`
* `parse_timestamp` (tries `"%Y-%m-%dT%H:%M:%S%.f"`, then `" %H:%M:%S%.f"`, then bare date as midnight, then converts via `and_utc` to seconds, millis, micros, or nanos depending on `TimeUnit`) for `Timestamp`

`Value::Int64` into `Float64` widens, into `String` stringifies, into `Dictionary` encodes. Other cross type cases become `None`.

## Dictionary

`Dictionary { codes: Vec<Option<i32>>, dict: Vec<String>, index: HashMap<String,i32> }`

* `dict_code` does `if let Some(&code) = index.get(v) { return code }` else `dict.push`, `index.insert`.

* `extend_owned` for `Dictionary` remaps the right dictionary into the left in one pass via `remap: Vec<i32> = b_dict.iter().map(|val| dict_code(a_dict, a_index, val)).collect()` then translates `a_codes.extend(b_codes.iter().map(|c| c.map(|idx| remap[idx as usize])))`.

* `try_upgrade_to_dict` is described in [Columnar](./columnar.md) and [Engine](./engine.md).

## Unification and promotion

`unify_variants` and `promote_to_variant` are the only places that change storage type. `merge.rs:extend` and `engines_to_record_batches` call `unify_variants(skey, okey)` before `extend_owned`; if `None`, they return `Error::Merge("column '{name}' has conflicting types ({skey} vs {okey}); provide explicit field_types")`. Promotions are `Int64 -> Float64` via `take` plus `map as f64`, and `String -> Dictionary` via rebuilding `dict`/`index`/`codes`. Same key is a no op.

## Finish

`TableBuilder::finish` (also `ColumnarSink::finish`) does:

1. `normalize` (truncate `len > row_count` and clear `row_dirty`)
2. Early `new_empty` if `column_order` empty
3. `auto_dict_upgrade` (if `plan.auto_dict`)
4. `sort_columns` by `schema_order`
5. Build `fields` plus `arrays` by iterating `column_order` and `get_column(name)` to call `arrow_datatype` and `to_arrow_array`
6. `Schema::new(fields)` plus `RecordBatch::try_new`

No borrowed bytes outlive `finish`; `StrColumn` owns its data and numeric Vecs are owned.

## Compare filter reapplication

`arrow_export::apply_compare_filter` exists for callers filtering an already built `RecordBatch` (for example merging). It is *not* the per row filter. Per row `FilterPredicate::check` is authoritative; `apply_compare_filter` only reapplies pure `Compare` and `And` of `Compare` via Arrow compute (`compare_columns` casts both to `Float64` if numeric else `Utf8`, then `gt`, `lt`, `gt_eq`, `lt_eq`, `eq`, `neq`, and `and` for `And`, plus `filter_record_batch`). Trees containing `Or`, `Not`, `Equal`, or `NotEqual` are returned unchanged to avoid double null semantics. This matches the comment in `arrow_export.rs:24`.

## Error mapping

`StrColumn::to_arrow` can return `ArrowError` (offsets, null buffer). `to_arrow_array` for dictionary can return `ArrowError` for `DictionaryArray::try_new`. Both surface as `Error::Arrow` and, via `rypipe-python`, as `PyException` with `Arrow error: ...`. `Utf8` from `simdutf8` surfaces as `ParseError` in Python.

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

# Real-world adapter: crxml

`crxml` is a high-throughput adapter for Crystal Reports XML exports. It is a good example of what a production `rypipe` adapter looks like: a small Rust crate that implements `rypipe-core`'s `Splitter` and `RecordParser` traits, plus a thin Python layer that registers the adapter with `rypipe`.

## What it does

Crystal Reports exports tabular data inside XML elements like `<Field Name="X"><Value>123</Value></Field>` and `<Text Name="Y"><TextValue>abc</TextValue></Text>`. `crxml` reads these exports and turns them into Arrow tables or DataFrames.

On the same workstation used for the rypipe engine benchmarks (AMD Ryzen 9 5900X 3.8 GHz, Arch Linux, 5800X 8C/16T measured), `crxml` parses Crystal Reports XML at **2.6–3.0 GB/s parallel** (1 GB, 926k `Details` rows, `par32` 2994 MB/s, `par16` 2720 MB/s, `par8` 2485 MB/s) and **714 MB/s single** (`read_to_columnar`), all warm-cache best-of-3. Streaming (`CrxmlReader` `lib.rs:534`, now also scanner-based via `RowSink` `lib.rs:564`) is **508 MB/s** 100 MB / **498 MB/s** 1 GB (was 251/234 `quick-xml`), within 30% of columnar (was 174% gap). The 1 GB `drop_all` pushdown reaches **4183 MB/s** parallel (CPU-bound, not I/O — `cat` 33 GB/s, `prefault` only +6%). That number is parser-bound; the `rypipe-core` engine (`Vec<ColumnBuilder>`+`field_index` `engine.rs:16`, `row_dirty` `engine.rs:26`) keeps up without being the bottleneck.

## How it fits into rypipe

Legend: `(ADAPTER BOUND)` lives in the adapter crate (`crxml`); `(CORE)` lives in `rypipe`.

```text
Crystal Reports XML file  (ADAPTER BOUND) input (row_tag = "Details" etc.)
    |
    v
CrystalXmlSplitter  : finds row-tag boundaries          (ADAPTER BOUND)  rypipe_core::Splitter impl (crxml-core/src/xml/splitter.rs)
    |
    v  Vec<Range<usize>>  (CORE) helper
CrystalXmlDecoder   : extracts fields from each row     (ADAPTER BOUND)  rypipe_core::RecordParser impl (crxml-core/src/xml/decoder.rs)
    |  Value events (Str, Int64, ...)  (CORE) enum
    v
rypipe-core engine  : typed builders, filters, projection, Arrow export  (CORE)  rypipe-core/src/engine.rs plus columnar.rs plus plan.rs
    |  RecordBatch  (CORE) Arrow
    v
pyarrow.Table / pandas.DataFrame  [PYTHON CORE] via rypipe-python C Data Interface plus (ADAPTER BOUND) registration (crxml/rypipe_adapter.py)
```

The Rust side lives in `crxml-core`. The Python side is a small `CrystalXMLAdapter` that calls the Rust core and registers itself with `rypipe`.

## Rust implementation

### Splitter

`crxml` implements `rypipe_core::Splitter` in `src/xml/splitter.rs` (also shared with the scanner via `find_special_regions` `splitter.rs:61` and `next_row_start` `splitter.rs:107`). Its job is to find safe chunk boundaries for parallel parsing.

Key ideas:

- Scan for the row tag with `memchr::memmem`, which is SIMD-accelerated on most platforms.
- Skip comment (`<!-- ... -->`) and CDATA (`<![CDATA[ ... ]]>`) regions so a `<Row` string inside them is not mistaken for a real row start.
- Validate that a candidate tag is followed by whitespace, `>`, or `/` to avoid prefix collisions such as `<RowItem`.
- Estimate bytes per row from a 64 KiB sample so the scheduler can size chunks and memory budgets (`TableBuilder::with_plan` `lib.rs:275` `cap = len/est_row`).

```rust
use rypipe_core::Splitter;
use memchr;

pub struct CrystalXmlSplitter { row_tag: Vec<u8> }

impl Splitter for CrystalXmlSplitter {
    fn find_split_points(&self, bytes: &[u8], max_chunks: usize) -> Vec<usize> {
        let ranges = compute_splits(bytes, &self.row_tag, max_chunks);
        let mut points = vec![0];
        for r in ranges { points.push(r.end); }
        points.dedup();
        if points.last() != Some(&bytes.len()) { points.push(bytes.len()); }
        points
    }

    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
        let sample_end = sample.len().min(65536);
        let row_tag_count = memchr::memmem::find_iter(&sample[..sample_end], &self.row_tag).count();
        sample_end
            .checked_div(row_tag_count)
            .unwrap_or(512)
            .max(1)
    }
}
```

### Record parser

`CrystalXmlDecoder` implements `rypipe_core::RecordParser` in `src/xml/decoder.rs`. Since `crxml 1.2` it uses a hand-rolled `memchr`/`memmem` scanner `src/xml/scanner.rs` (not `quick_xml`) — the same scanner backs both the columnar and the super-optimized streaming path (`lib.rs:534` `RowParser`+`RowSink`+`scan_one_row` `scanner.rs:81`).

For each row element it:

1. Emits row attributes as fields (`emit_all_attrs` `scanner.rs:180`).
2. Walks child elements via `scan_child` `scanner.rs:121` (`next_lt` `scanner.rs:412` `memchr(b'<')` + `in_region` `scanner.rs:418` fast-path for `find_special_regions`).
3. Recognizes `<Field Name="..."><Value>...</Value></Field>`, `<Text Name="..."><TextValue>...</TextValue></Text>`, and `<Section SectionNumber="..."/>` via `field_element` `scanner.rs:202`, `text_element` `scanner.rs:280`, `section_element` `scanner.rs:352`.
4. Byte-jumps dropped fields: `if !sink.wants(&key)` `scanner.rs:210` → `find_close_after` with `LazyLock<Finder>` `scanner.rs:569` for `</Field>` etc., without visiting `<Value>` children.
5. Calls `sink.put_field(key, Value::Str(value))` (`row` attrs `scanner.rs:180`) or `wants`+`put_field` for `Field`/`Text` so the engine builds typed columns. The `resolve`/`put_field_resolved` pair `decoder.rs:46`/`53` is available for expensive extraction (one hash instead of two), used for `Section`/unknown.

A simplified excerpt (columnar):

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

pub struct CrystalXmlDecoder { row_tag: Vec<u8> }

impl RecordParser for CrystalXmlDecoder {
    fn validate(&self, bytes: &[u8]) -> Result<()> {
        simdutf8::basic::from_utf8(bytes)?;
        Ok(())
    }

    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        crate::xml::scanner::scan_chunk(bytes, &self.row_tag, sink)
    }
}
```

Streaming reuses the same scanner row-by-row:

```rust
// lib.rs:534 RowParser with InputBuffer (mmap) + RowSink
let mut sink = RowSink { row: &mut self.row };
crate::xml::scanner::scan_one_row(bytes, self.pos, &row_tag, &regions, &mut sink);
```

The scanner is `wants`-driven and region-aware, so chunked and streaming stay correct without a serial pre-pass.

## Python adapter registration

`crxml` exposes a `CrystalXMLAdapter` that wraps `CrystalXMLSource` and registers it with `rypipe`:

```python
import rypipe
from crxml.source import CrystalXMLSource

class CrystalXMLAdapter:
    def read(self, path: str, **kwargs):
        return CrystalXMLSource(path, **kwargs).to_arrow()

rypipe.register_adapter("crxml", CrystalXMLAdapter(), extensions=[".xml"])
```

Importing `crxml` now makes the adapter available automatically:

```python
import rypipe

table = rypipe.read("report.xml", format="crxml", row_tag="Row")
```

The same `row_tag`, `field_types`, `filter`, `memory`, and `chunks` options from `CrystalXMLSource` are passed through, so users get the full engine feature set through the generic `rypipe` API.

## Why it is fast

| technique | benefit |
|-----------|---------|
| Hand-rolled `memchr`/`memmem` scanner `scanner.rs:29` (`memchr`, `memchr3`, `Finder` `scanner.rs:569`) | No `quick-xml` event loop (was 42% wall `lib.rs:598`); SIMD `scan_open_tag` `scanner.rs:414`, `field_element` `scanner.rs:202`, `Find-er` byte-jump for dropped `Field` (`wants` `scanner.rs:210` → `find_close_after`) |
| `memchr::memmem` row-tag scan + skip-region | SIMD `next_row_start` `splitter.rs:107`, `find_special_regions` `splitter.rs:61` (`<!--`, `<![CDATA[`) with `is_empty` fast path `scanner.rs:418` |
| SIMD UTF-8 validation | `simdutf8` `decoder.rs:32` validates chunk bulk; `utf8_unchecked` `scanner.rs:39` + conditional `unescape` only if `b'&'` `scanner.rs:662` |
| `rypipe-core` `Vec<ColumnBuilder>`+`field_index` `engine.rs:16` single `get` + `row_dirty` `engine.rs:26` bitmask | `push_field` was double `HashMap` probe → single `field_index.get` → `columns[idx]`; `finish_row` now only null-fills clear bits, not all `C` columns |
| `InputBuffer` `mmap` `lib.rs:25` `auto_mmap` >50 MB + `cap` via `estimate_bytes_per_row` `splitter.rs:41` | Avoids `fs::read` `rep_movs` 3% + over-reserve `Vec` 2× |
| Parallel fast path `parallel.rs:82` + streaming `RowSink` `lib.rs:564` | `auto_dict`/`Compare` off → per-chunk `RecordBatch` export in parallel; streaming reuses same scanner row-by-row via `scan_one_row` `scanner.rs:81` without `TableBuilder` |

## Lessons for adapter authors (from `crxml` super-optimization)

1. **Specialize the parser**: generic line splitting is fine, but real throughput comes from a format-aware `memchr` scanner (`crxml` went `quick-xml` 251 MB/s → scanner 508 MB/s streaming, 489→714 columnar single).
2. **Find split points cheaply**: one `memmem` `splitter.rs:27` + `find_special_regions` `splitter.rs:61` with `is_empty` fast path beats byte-by-byte; `estimate_bytes_per_row` `splitter.rs:41` sizes `TableBuilder` `lib.rs:275`.
3. **Handle boundary cases**: chunks can start inside a row; `scan_one_row` `scanner.rs:81` (`Recover` → `pos+1`) and `wants`-driven `find_close_after` keep parallel correct.
4. **Borrow strings into the engine**: `Value::Str(&str)` slices via `utf8_unchecked` `scanner.rs:39` + conditional `&` `scanner.rs:662` avoids `Cow` alloc (94% of values are plain ASCII). `RowSink` `lib.rs:564` pushes directly without `TableBuilder` hash/arena for streaming.
5. **Skip dropped fields in the scanner**: check `wants`/`resolve` *before* visiting `<Value>` children — `field_element` `scanner.rs:210` byte-jumps to `</Field>` via `Finder` (drop_all 4183 MB/s, 66% win).
6. **Reuse the scanner for both engines**: columnar `scan_chunk` `scanner.rs:54` and streaming `scan_one_row` `scanner.rs:81` share `parse_row` `scanner.rs:73`, so one optimization benefits `stream`/`columnar`/`parallel`/`bounded`.
7. **Measure first**: `perf` `scan_open_tag` 8.3% + `field_element` 8.6% + `push_field_resolved` 2.76% vs `rep_movs` 3% tells you `mmap` is 3% but scanner is 35% — focus there. `benchmarks/bench_extended.py` (104 benchmarks/file) covers all engines×sinks×pushdowns×chunk/bounded/batch/pipeline.
8. **Register with `rypipe`**: thin `CrystalXMLAdapter` keeps `rypipe.read(format="crxml")` while Rust stays fast.

## Source

The full implementation is in the [crxml repository](https://github.com/emiliano-go/crxml), especially:

- `src/crxml_core/src/xml/splitter.rs`
- `src/crxml_core/src/xml/decoder.rs`
- `src/crxml/rypipe_adapter.py`

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

# rypipe examples

This page shows common patterns in Python and Rust. All Python examples assume a registered adapter is installed (for example `pip install crxml` for XML). `rypipe` itself does not ship format parsers.

Legend: `(ADAPTER BOUND)` is code you write in your adapter crate (format specific). `(CORE)` is code in `rypipe` crates (reused). The split is the same as in [Architecture](./architecture/).

## Python API examples

### Read a file through a registered adapter

```python
import rypipe

table = rypipe.read("data.xml", format="crxml", row_tag="Row")
print(table.num_rows, table.num_columns)
```

### Chain a pipeline

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

df = (
    rypipe.read("data.xml", format="crxml", row_tag="Row")
    | RenameFields({"old_name": "new_name"})
    | DropFields(["internal_id"])
    | FilterRows(field="status", op="==", value="active")
    | CastTypes({"amount": "float64"})
).to_dataframe()
```

### Build an adapter subclass

```python
import rypipe
import pyarrow as pa

class DummyAdapter(rypipe.Adapter):
    def read(self, path, **kwargs):
        # A real adapter would call a Rust parser here.
        return pa.table({
            "id": [1, 2, 3],
            "name": ["alice", "bob", "carol"],
        })

# Register once per session
rypipe.register_adapter("dummy", DummyAdapter, extensions=[".dummy"])

table = rypipe.read("example.dummy")
```

### Work with a Source object directly

```python
from crxml import CrystalXMLSource

source = CrystalXMLSource("report.xml", row_tag="Row")

# Iterate rows without loading the whole table
for row in source:
    print(row["id"])

# Or export to Arrow and reuse it
arrow = source.to_arrow()
df = source.to_dataframe()
source.to_parquet("report.parquet")
```

### Memory-bounded stream

```python
import rypipe

# Adapter decides how to honor the memory budget.
stream = rypipe.read_stream("huge.xml", format="crxml", memory="256MiB")
for batch in stream.to_batches(max_chunksize=4096):
    process(batch.to_pylist())
```

### Parallel parse

```python
import rypipe

table = rypipe.read_par("large.xml", format="crxml", chunks=16)
```

### Sink to Parquet

```python
import rypipe
from rypipe import to_parquet

pipeline = (
    rypipe.read("data.xml", format="crxml", row_tag="Row")
    | FilterRows(field="status", op="==", value="active")
)
to_parquet(pipeline, "active.parquet")
```

## Rust API examples

All Rust examples use the `rypipe-core` crate. `(ADAPTER BOUND)` parts are the `Splitter` and `RecordParser` you implement; `(CORE)` parts are the engine.

### Define a custom Splitter and RecordParser  (ADAPTER BOUND)

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

#[derive(Clone, Default)]
pub struct LogSplitter; // (ADAPTER BOUND)

impl Splitter for LogSplitter { // (ADAPTER BOUND)
    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];
        for (i, &b) in bytes.iter().enumerate().skip(1) {
            if b == b'\n' && points.len() < max_chunks { points.push(i + 1); }
        }
        if *points.last().unwrap() != bytes.len() { points.push(bytes.len()); }
        points
    }
    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
        (sample.len() / sample.iter().filter(|&&b| b == b'\n').count().max(1)).max(1)
    }
}

#[derive(Clone, Default)]
pub struct LogParser; // (ADAPTER BOUND)

impl RecordParser for LogParser { // (ADAPTER BOUND)
    fn validate(&self, bytes: &[u8]) -> Result<()> { // (ADAPTER BOUND) check UTF-8
        simdutf8::basic::from_utf8(bytes).map_err(rypipe_core::Error::Utf8)?;
        Ok(())
    }
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> { // (ADAPTER BOUND) emits (CORE) Value events
        let text = std::str::from_utf8(bytes).unwrap();
        for line in text.lines() {
            if line.is_empty() { continue; }
            sink.begin_row(); // (CORE) TableBuilder
            for part in line.split(',') {
                if let Some((k, v)) = part.split_once('=') {
                    if let Some(r) = sink.resolve(k) { // (CORE) one hash
                        sink.put_field_resolved(r, Value::Str(v)); // (CORE) single index lookup plus dirty
                    }
                }
            }
            sink.end_row(); // (CORE) null fill plus filter
        }
        Ok(())
    }
}
```

### Run single-threaded, parallel, and bounded  (CORE) drivers

```rust
use rypipe_core::{ExecutionPlan, FieldType, Pipeline, MemoryBudget}; // (CORE)
use std::path::Path; // (CORE) InputBuffer uses Path

// (CORE) Pipeline wires (ADAPTER BOUND) splitter/parser to (CORE) engine
let pipeline = Pipeline::new(LogSplitter, LogParser).with_plan( // (CORE) plus (ADAPTER BOUND) types
    ExecutionPlan::new().type_as("amount", FieldType::Float64).drop("internal_id")
);

// (CORE) Single thread, whole file
let batch = pipeline.read_path(Path::new("data.log"), false, false)?; // (CORE) via InputBuffer

// (CORE) Parallel (rayon), chunked, fast path when !auto_dict and schemas consistent
let batches = pipeline.read_path_par(Path::new("data.log"), 8, false, false)?; // (CORE)

// (CORE) Bounded memory (streaming), from file or from bytes
let batches = pipeline.read_path_stream(Path::new("huge.log"), MemoryBudget::new(128 * 1024 * 1024), false)?; // (CORE)
let batches = pipeline.read_bytes_stream(data_bytes, MemoryBudget::new(64 * 1024 * 1024))?; // (CORE) no file IO, slicing
```

### Low level direct TableBuilder  (CORE)

```rust
use rypipe_core::{ExecutionPlan, TableBuilder, ColumnarSink, Value}; // (CORE)
use std::path::Path; // (CORE)

let plan = ExecutionPlan::new().drop("internal_id"); // (CORE)
let input = rypipe_core::InputBuffer::open(Path::new("data.log"), false, false)?; // (CORE) handles mmap plus gzip/zstd/lz4
let mut builder = TableBuilder::with_plan(1024, plan); // (CORE)
LogParser.validate(input.as_slice())?; // (ADAPTER BOUND)
LogParser.parse_chunk(input.as_slice(), &mut builder)?; // (ADAPTER BOUND) -> (CORE) sink
let batch = builder.finish()?; // (CORE) normalize, auto_dict, sort, to_arrow
```

### Export a RecordBatch to PyArrow  (CORE) helper plus (ADAPTER BOUND) registration

```rust
use rypipe_python::{execution_plan_from_kwargs, record_batches_to_pyarrow_table}; // (CORE) helper
use pyo3::prelude::*; // (CORE) plus (ADAPTER BOUND) glue

#[pyfunction]
fn read_log(py: Python, path: &str, field_mapping: Option<std::collections::HashMap<String,String>>) -> PyResult<pyo3::Bound<pyo3::PyAny>> {
    let plan = execution_plan_from_kwargs(field_mapping, None, None, None, None, None, false, None, None)?; // (CORE)
    let batches = py.allow_threads(|| { // (CORE) GIL release
        use rypipe_core::{Pipeline, MemoryBudget};
        Pipeline::new(LogSplitter, LogParser).with_plan(plan).read_path_par(path, 4, false, false) // (CORE) plus (ADAPTER BOUND)
    })?;
    record_batches_to_pyarrow_table(py, &batches) // (CORE) C Data Interface
}
```

## Adapter package layout

A minimal adapter package looks like this:

```
rypipe-foo/
├── Cargo.toml          # depends on rypipe-core + rypipe-python
├── pyproject.toml      # maturin build
├── src/
│   └── lib.rs          # Rust parser + PyO3 module
└── rypipe_foo/
    └── __init__.py     # Python adapter + rypipe.register_adapter
```

The Python side:

```python
import rypipe

class FooAdapter(rypipe.Adapter):
    def read(self, path, **kwargs):
        return _rypipe_foo.read_foo(path, **kwargs)

rypipe.register_adapter("foo", FooAdapter, extensions=[".foo"])
```

See [Writing adapters](writing-adapters.md) for the full trait reference.

========================================================================
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/): the generic engine (see [Architecture overview](./architecture/) for crate map).
  - [`rypipe-python`](./architecture/#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/): how the pieces fit together (start with [Overview](./architecture/index.md), then [Engine](./architecture/engine.md), [Columnar](./architecture/columnar.md), [Plan](./architecture/plan.md), [Execution](./architecture/execution.md), [Data flow](./architecture/data-flow.md), [Storage](./architecture/storage.md), [Optimizations](./architecture/optimizations.md)).
- [Why Python?](./why-python.md): why rypipe is Rust core plus Python surface, not pure Rust : the data driven case for the hybrid.
- [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 | RSS |
|------|------|------|--------|------|-----|
| `Pipeline::read_path` | 5,000,000 | 1.31 s | 3.83 M | 229 | 290 MB |
| `Pipeline::read_path_par` (4 chunks) | 5,000,000 | 1.46 s | 3.43 M | 205 | 432 MB |
| `Pipeline::read_path_par` (8 chunks) | 5,000,000 | 1.49 s | 3.35 M | 200 | 529 MB |
| `Pipeline::read_path_stream` (64 MiB) | 5,000,000 | 1.98 s | 2.53 M | 151 | 629 MB |

Single-thread parse is fastest for this simple TSV adapter because chunk
overhead dominates. Parallel mode trades a small throughput drop for much
higher CPU utilization and scales better with heavier parsers. The bounded
streaming path keeps intermediate batches near the 64 MiB budget but has higher
peak RSS because of the Arrow export buffer.

Run the benchmark yourself:

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

# Python wrapper that also writes JSON results
python benchmarks/bench_throughput.py --output .benchmarks/rypipe.json
```

## 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.

## Example: `crxml` adapter

`crxml` (Crystal Reports XML, at `docs/crxml-adapter.md`) is the reference for what a good adapter looks like: hand-rolled `memchr` scanner `crxml-core/src/xml/scanner.rs` instead of `quick-xml`, shared between columnar `scan_chunk` and streaming `scan_one_row` + `RowSink` `crxml-core/src/lib.rs:564`, `Vec<ColumnBuilder>`+`field_index` + `row_dirty` bitmask in `engine.rs:16`, `InputBuffer` `mmap` auto for >50 MB (`crxml-core` `auto_mmap`), and `wants`-driven skip-bytes. On a Ryzen 5800X it holds **714 MB/s single / 2.6–3.0 GB/s parallel** (1 GB, 926k rows) and **4183 MB/s** with `drop_all` pushdown, with streaming now **508 MB/s** (2× over `quick-xml`'s 251). See `benchmarks/bench_extended.py` (104 benchmarks/file) for the full matrix (native, source×sink, pushdowns, chunk/bounded/batch/pipeline).

## 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/): 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 fusable pipeline stages. 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: `==`, `!=`, `>`, `<`, `>=`, `<=`.

Boolean combinators (`FilterRowsAny`, `FilterRowsAll`, `FilterRowsNot`) are
fusable and build predicate trees that are evaluated per-row inside the Rust
parse loop:

```python
from rypipe import FilterRowsAny, FilterRowsAll, FilterRowsNot

# OR, AND, NOT (keyword-form leaves only; callables are not fusable)
FilterRowsAny(
    FilterRows(field="status", op="==", value="active"),
    FilterRows(field_a="age", op=">=", field_b="min_age"),
)
FilterRowsAll(
    FilterRows(field="region", op="==", value="us"),
    FilterRows(field="status", op="==", value="active"),
)
FilterRowsNot(FilterRows(field="flag", op="==", value="deleted"))

# Chaining FilterRows stages is an implicit AND:
pipeline = src | FilterRows(field="a", op="==", value="1") | FilterRows(field="b", op="==", value="2")
# equivalent to FilterRowsAll(...)

# Arbitrary nesting via the filter dict:
filter={"or": [{"field": "status", "op": "==", "value": "active"}, {"not": {"field": "flag", "op": "==", "value": "deleted"}}]}
filter={"and": [{"field": "a", "op": "==", "value": "1"}, {"field_a": "x", "op": ">", "field_b": "y"}]}
```

### 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`;
`record_batches_to_pyarrow_batches` exports them as a list of individual
`pyarrow.RecordBatch` objects for streaming-style APIs.

## 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"`, `"string"`, `"date32"`, or `"timestamp"` (also `"timestamp[s]"`, `"timestamp[ms]"`, `"timestamp[us]"`, `"timestamp[ns]"`). |
| `dictionary` / `dictionary_columns` | `list[str]` | Explicit dictionary encoding. |
| `filter` | `dict` | Per-row filter (see below). |
| `schema` | `list[str]` | Output column order. |
| `auto_dict` | `bool` | Upgrade low-cardinality string columns to dictionary. |
| `auto_dict_threshold` | `float` | Max distinct/row ratio for auto-dict (default `0.05`). |
| `auto_dict_max_size` | `int` | Max dictionary entries for auto-dict (default `256`). |
| `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 per-row during parse with native-typed
comparison; numeric columns promote Int64/Float64):

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

Supported comparison ops: `>`, `<`, `>=`, `<=`, `==`, `!=`. Mismatched
non-numeric types or null operands fail the row.

Boolean combinators use the same leaf shapes above with
`{"and": [spec, ...]}`, `{"or": [spec, ...]}`, and `{"not": spec}` and may be
nested arbitrarily; evaluation is per-row with short-circuiting, so missing-
field and type-mismatch behaviour of leaves is preserved (a negated missing
field, for example, keeps the row).

## Streaming batches

```python
import rypipe

# Bounded-memory read yielding pyarrow.RecordBatch objects.
for batch in rypipe.read_batches("huge.csv", memory="256MiB"):
    process(batch)

# Same, from a Source or Pipeline.
src = MySource("huge.csv")
for batch in src.iter_arrow_batches(batch_size=10_000):
    process(batch)
```

`read_batches` and `iter_arrow_batches` yield Arrow record batches one at a
time. Memory is bounded during parsing; batches are currently produced from a
materialized table, so peak memory is not yet fully incremental.

## 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/): 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::{CsvDecoder, CsvSplitter};

fn main() -> rypipe_core::Result<()> {
    let pipeline = Pipeline::new(CsvSplitter, CsvDecoder {
        header: vec!["id".into(), "status".into(), "amount".into()],
    })
        .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,
    )?;

    // In-memory byte slices (useful for BytesIO, stdin buffers, or decompressed
    // content) without file I/O.
    let bytes = std::fs::read("data.csv")?;
    let batch = pipeline.read_bytes(&bytes)?;
    let batches = pipeline.read_bytes_par(&bytes, 8)?;
    let batches =
        pipeline.read_bytes_stream(&bytes, rypipe_core::MemoryBudget::new(500_000_000))?;

    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::{CsvDecoder, CsvSplitter}; // separate adapter crate
use std::path::Path;

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

    let decoder = CsvDecoder {
        header: vec!["id".into(), "status".into(), "amount".into()],
    };
    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")
    // leaf filters; combine with And/Or/Not trees (see below)
    .filter_eq("status", "active");

// Filter trees from ExecutionPlan helpers or directly:

let any = FilterPredicate::any(
    FilterPredicate::Equal { field: "a".into(), value: "1".into() },
    FilterPredicate::not(FilterPredicate::Equal { field: "b".into(), value: "2".into() }),
);
let all = FilterPredicate::all(
    FilterPredicate::Compare { field_a: "amount".into(), op: CompareOp::Gt, field_b: "threshold".into() },
    FilterPredicate::Equal { field: "status".into(), value: "active".into() },
);

let plan_with_tree = ExecutionPlan {
    filter: Some(FilterPredicate::any(any, all)),
    field_types: plan.field_types.clone(),
    ..plan.clone()
};
```

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::{CsvDecoder, CsvSplitter}; // separate adapter crate

let bytes = std::fs::read("data.csv")?;
let splitter = CsvSplitter;
let decoder = CsvDecoder {
    header: vec!["id".into(), "status".into(), "amount".into()],
};
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 with a unified schema; the merge path returns a single merged
batch when `auto_dict` is enabled or chunks disagree irreconcilably on column
types.

## Bounded parse (low level)

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

let splitter = CsvSplitter;
let decoder = CsvDecoder {
    header: vec!["id".into(), "status".into(), "amount".into()],
};
let batches = BoundedExecutor::new(MemoryBudget::new(500_000_000))
    .run(Path::new("huge.csv"), &splitter, decoder, ExecutionPlan::new(), false)?;

// From an in-memory buffer (transparent decompression in InputBuffer::open
// also lands here when the file was compressed; see Cargo features below).
let bytes = std::fs::read("huge.csv")?;
let batches = BoundedExecutor::new(MemoryBudget::new(500_000_000))
    .run_bytes(&bytes, &splitter, decoder, ExecutionPlan::new())?;
```

## Transparent compressed-file decoding

File inputs are auto-detected by magic bytes and transparently decompressed
when the corresponding Cargo feature is enabled:

```toml
[dependencies]
rypipe-core = { version = "0.1", features = ["gzip"] } # or "zstd", "lz4", "compress-all"
```

Supported magics: gzip `1f 8b`, zstd `28 b5 2f fd`, lz4 frame `04 22 4d 18`.
Detection happens in `InputBuffer::open`; decompressed bytes are served from
memory (`InputBuffer::Owned`) across all execution modes.

## Apply a Compare filter to an existing batch

Pure column-comparison filters (``Compare`` and ``And`` of ``Compare``) are
normally evaluated per-row during parsing but can also be applied to an
already-built ``RecordBatch``. Trees involving ``Or``, ``Not``, ``Equal``, or
``NotEqual`` are no-ops at this layer (per-row evaluation is already
authoritative) so valid rows can never be dropped twice. To filter an
already-built ``RecordBatch`` use
``apply_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_batch_to_pyarrow,
    record_batches_to_pyarrow_batches, record_batches_to_pyarrow_table,
};

// Inside a PyO3 function:
let plan = execution_plan_from_kwargs(...)?;
let table = record_batches_to_pyarrow_table(py, &batches)?;
// Or export batches individually for streaming-style APIs:
let batch_list = record_batches_to_pyarrow_batches(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/): how the pieces fit together.
- [Python API](./python-api.md): the Python bindings over the same engine.

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

# Why Python, not pure Rust?

`rypipe` is *Rust where it counts, Python where it ships*. The core is pure Rust for speed and memory safety; the surface is Python for reach and productivity. This page explains why that split is intentional : not a compromise.

## TL;DR

* **Data work lives in Python.** Notebooks, ETL glue, and the modern lakehouse are Python-first.
* **Rust wins the hot path, Python wins composition.** Parsing 100 M fields at 700 MB/s needs Rust; chaining `RenameFields | FilterRows | CastTypes → .to_pandas()` needs Python.
* **Arrow is the bridge.** `rypipe` produces Arrow in Rust and hands it to `pyarrow`/`pandas`/`Polars` with no copy via the C Data Interface : the same columnar substrate every data tool speaks.
* **Pure Rust would shrink the audience 10×.** ETL and analysis are done by data engineers, analysts, and ML engineers : most never touch `cargo build`.

If you need Rust-only, `rypipe-core` is still there (see [Rust API](./rust-api.md)). The Python layer is a *thin* orchestration shell, not a bottleneck.

---

## 1. Data is Python-first : by every measure

| Signal | What it says |
|--------|--------------|
| **Stack Overflow Developer Survey 2024** : 65 000 responses | Python is the 3rd most-used language overall (51% of professional developers), #1 for beginners, and has been top-3 for 8 years straight. |
| **Kaggle AI & Data Science Survey 2023** : 20 000 practitioners | 84% use Python regularly; R is second at 32%. For ETL + analysis + modeling, Python is the *lingua franca*. |
| **JetBrains State of Developer Ecosystem 2023** | 57% of data engineers write Python daily; Rust is at 12% overall and <3% in data roles. |
| **PyPI vs crates.io** | `pandas` alone has 40 M monthly downloads; the top 20 Rust data crates combined are two orders of magnitude lower. |
| **Job market** | LinkedIn “data engineer” in US/EU: 9/10 listings require Python, 1/50 mention Rust. |

The takeaway is not that Rust is bad : it is excellent for the engine : but that the *consumer* of an ingestion engine is almost always a Python program.

## 2. ETL is glue : and Python is the best glue

Real ETL is rarely “parse one file as fast as possible.” It is:

```
S3 / API / FTP → decompress → parse → rename/drop/cast/filter → validate → join/aggregate in DuckDB/Spark → write Parquet → publish to warehouse
```

* **Orchestration** is Python (`Airflow`, `Dagster`, `Prefect`, `Mage`, plain `cron`). A DAG that can `import rypipe` keeps the whole pipeline in one language.
* **File handling** is Python (`pathlib`, `fsspec`, `boto3`, `requests`). Requiring users to shell out to a Rust binary breaks composability.
* **Interactivity** is Python. Data is discovered in a Jupyter notebook: `source | FilterRows(...) | .to_dataframe().describe()`. The same code then runs headless in CI. A pure-Rust library forces a context switch (new language, new toolchain, new mental model) exactly when exploration should be fastest.
* **Error handling & data quality** are Python (`pydantic`, `pandera`, `great_expectations`). Row-level validation, Slack alerts, and quarantine logic live next to the parse call : not inside the parser.

A pure-Rust `rypipe` would need users to learn `cargo`, `tokio`, lifetimes, and Arrow’s Rust API to do what today is one `pip install` and five lines.

## 3. The ecosystem gravity is Arrow + Python

Arrow’s design assumes language boundaries are crossed at the `RecordBatch` level:

```
Rust parser (rypipe-core) ──C Data Interface──► pyarrow.Table ──► pandas / Polars / DuckDB / Spark
```

* **`pyarrow`** is the de-facto Arrow implementation in Python (25 M downloads/month). `rypipe` hands batches via the C Data Interface : no serialization, no copy, no GIL.
* **`pandas 2.0+`** stores columns as `ArrowDtype` when `dtype_backend="pyarrow"` : rypipe tables become DataFrames for free.
* **`Polars`** is itself a Rust core with Python bindings; `pl.from_arrow(table)` is zero-copy. Users already understand “Rust engine, Python API.”
* **`DuckDB`, `DataFusion`, `DaFt`** all consume Arrow from Python. rypipe fits as the *ingestion* stage before the query engine, not a competitor.

A pure-Rust engine would still need to export Arrow : and then every downstream step would re-wrap it in Python anyway. Putting the binding in the engine removes that friction once.

## 4. Performance: Rust hot loop, Python orchestration

`rypipe`’s single-thread XML throughput on a 702 MB file is ~700 MB/s (95% in `parse`, 3% in `split`, 2% in `finish`). Parallel ×8 is ~2.5 GB/s. The profile for a real Crystal Reports export shows:

* `field_element` / `scan_open_tag` ~17% : XML scanning (Rust)
* `push_field_resolved` 2.76% + `field_index.get` 1.64% : column dispatch (Rust, Vec+map after 0.1.2)
* `rep_movs` 3–10% : the un-avoidable `file → page cache` copy

Python time is **orchestration only**: `ExecutionPlan` construction, `Pipeline` `|` chaining, and the final `Table.to_pandas()` conversion : microseconds per row. The GIL is released for the entire `Pipeline::read_*` call (`py.allow_threads` in `crates/rypipe-python`), so multi-core parsing is not gated.

In other words: *you pay Rust for the hot loop, Python for the composition*. A pure-Rust pipeline would move the same composition code into Rust and save ~0.1 ms : while losing the ability to `df.plot()`, `df.to_sql()`, or `requests.get()` next line.

## 5. What pure Rust would cost you

| Dimension | Hybrid (rypipe today) | Pure Rust alternative |
|-----------|----------------------|----------------------|
| **Install** | `pip install rypipe my-adapter` : one wheel, stable ABI (`abi3`, 3.10+), manylinux/musllinux/macOS/Windows | `cargo add rypipe-core my-adapter && cargo build --release` : requires Rust 1.78+, `cc`, Arrow build, per-project compilation |
| **Audience** | Anyone who can `import pandas` : data engineers, analysts, scientists | Rust developers only : <10% of data teams |
| **Prototyping** | `source \| FilterRows(...) \| .to_dataframe()` in a notebook cell, instant feedback | Write a binary, handle `Result`, print tables manually, rebuild on every change |
| **Reuse in prod** | Same notebook code runs in Airflow/Dagster unchanged | Rewrite notebook logic in Rust or maintain two codebases |
| **Tool chain** | Stays in `pip`/`conda`/`uv` : no new tool | Adds `rustup`, `cargo`, `miri`, `clippy` to every data repo |
| **Arrow interop** | `pyarrow`, `pandas`, `Polars` zero-copy out of the box | Must go through FFI or JSON/CSV round-trip to reach Python consumers anyway |
| **Adapter distribution** | Separate pip packages (`rypipe-csv`, `rypipe-json`) : `pip install` discovers them | Separate crates : `cargo add` per project, no central registry for data users |

## 6. When Rust-only *does* make sense : and rypipe still supports it

There are legitimate reasons to stay in Rust:

* You are building a Rust service (e.g., a sidecar that ingests files and writes Parquet to S3 without touching Python).
* You need to embed parsing in an existing Rust data plane (e.g., inside `DataFusion` or a custom `arrow` server).
* You want the smallest possible binary and no Python runtime.

For those cases, **use `rypipe-core` directly** : no Python required:

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

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

`rypipe-python` is an *additional* crate, not a replacement. The engine never depends on Python : you can publish a Rust-only adapter and depend on `rypipe-core = "0.1"` alone.

## 7. Design principle: data-driven development

Modern data work is **data-driven development**: the shape of the output dictates the shape of the code, not the other way around.

1. **See the data** (`source.schema()`, `source.to_pandas().head()`).
2. **Shape it interactively** (`RenameFields`, `FilterRows`, `CastTypes` with instant `.to_dataframe()` preview).
3. **Freeze the pipeline** (commit the `ExecutionPlan`).
4. **Scale it** (same pipeline via `read_stream` with `memory="1GiB"` on 100 GB of files).

Steps 1–2 demand a REPL. Step 4 demands Rust speed. rypipe gives both from the same object. A pure-Rust engine would optimize step 4 by pessimizing steps 1–3 : the 80% of time where humans, not CPUs, are the bottleneck.

> **Analogy:** `polars` didn’t win by being “pure Rust.” It won by putting a Rust engine behind `pl.DataFrame` : a Python/Rust hybrid that feels like pandas but runs at Rust speed. `rypipe` does the same for ingestion.

## 8. FAQ

**“Why not just use `csv` + `pandas.read_csv` / `polars.read_csv`?”**
For CSV alone you can. rypipe’s value is the *same* engine for XML, JSON, HTML, and other row-oriented formats where no fast Arrow-native reader exists : with parallel and memory-bounded execution, pushdown, and the same `source | ... | .to_polars()` surface.

**“Can I avoid Python entirely?”**
Yes. See [Rust API](./rust-api.md) and [Writing a format adapter](./writing-adapters.md). The Python bindings are optional; `rypipe-core` has zero Python dependency.

**“Does Python add overhead?”**
Measured: <1% of wall time for 90 k rows × 10 fields; 17 of 17 data-integrity tests assert bit-identical results across `read_bytes` (Rust), `read_bytes_par`, and `read_bytes_stream` via both APIs. See [Performance](./performance.md) for `bench_throughput`.

**“Will you maintain both APIs?”**
Yes. `rypipe-core` is versioned independently (`0.1.x`). `rypipe-python` follows `pyo3` (0.29) and `arrow` (59.2) and is tested on CPython 3.10–3.14. Breaking adapter APIs bumps the minor version.

---

**Bottom line:** data teams already live in Python. rypipe meets them there : with a Rust engine underneath so they never have to leave to go fast.

========================================================================
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.29", features = ["extension-module", "abi3-py310"] }
```

## 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 {
                // Point at the FIRST BYTE of the next record, not at the
                // newline itself; otherwise chunks begin with a stray '\n'.
                points.push(i + 1);
            }
        }
        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. Split points point at the
  first byte of a record, never at the delimiter itself.

## 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(rypipe_core::Error::Utf8)?;
        Ok(())
    }

    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        let text = simdutf8::basic::from_utf8(bytes).map_err(rypipe_core::Error::Utf8)?;
        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(())
    }

    // Fastest when extraction is expensive : one resolve instead of two:
    // (requires rypipe-core >=0.1.1)
    fn parse_chunk_fast(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        let text = simdutf8::basic::from_utf8(bytes).map_err(rypipe_core::Error::Utf8)?;
        for line in text.lines() {
            if line.is_empty() {
                continue;
            }
            sink.begin_row();
            for (col, value) in self.header.iter().zip(line.split(',')) {
                if let Some(resolved) = sink.resolve(col) {
                    // ... expensive unescaping / decoding of `value` here ...
                    sink.put_field_resolved(resolved, Value::Str(value));
                }
            }
            sink.end_row();
        }
        Ok(())
    }
}
```

Key points:

- Call `sink.wants(name)` before expensive extraction to skip dropped fields.
  When extraction itself is expensive (entity unescaping, base64, …), use the
  zero-copy pair `sink.resolve(name)` + `sink.put_field_resolved(resolved, value)`
  to pay the `rename→drop` hash only once instead of twice.
- 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.
- Columns are now stored as `Vec<ColumnBuilder>` with a `field_index: FxHashMap<String,usize>`
  beside `column_order`. The hot path `push_field` is a single index lookup
  (`field_index.get` → `columns[idx]`) instead of two `HashMap` probes; this
  benefits every adapter equally.

## 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>(
    py: Python<'py>,
    path: String,
    field_mapping: Option<HashMap<String, String>>,
    // ... other kwargs
) -> PyResult<Bound<'py, PyAny>> {
    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 `FilterRows` predicates (both
constant and column-to-column forms) 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, 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/): how splitters, parsers, and the engine interact.
- [Python API](./python-api.md): registering adapters with the `rypipe` package.
- [crxml adapter](./crxml-adapter.md): a production adapter that reaches 2.4 GB/s on Crystal Reports XML.