Stages¶

Stages are the building blocks of pipelines. Each stage transforms the data as it flows through. This page explains each stage and shows how to implement them in your adapter.

How stages work¶

Stages are the building blocks of pipelines. Each stage transforms the data as it flows through, like a Unix pipe:

from crxml import CrystalXMLSource, RenameFields, FilterRows

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

# Each stage transforms the data in order
result = (
    source
    | RenameFields({"Name": "name"})    # renames columns
    | FilterRows(field="Status", op="==", value="Active")  # keeps matching rows
)

table = result.to_arrow()

Under the hood, stages are callables that transform dicts. rypipe automatically pushes fusable stages (RenameFields, DropFields, CastTypes, FilterRows with keyword form or compiled lambda) into the Rust parse loop for maximum performance. Non-fusable stages (non-resolvable lambda predicates) run in Python. This is called plan fusion. See Plans for details.

RenameFields¶

Renames columns in each record.

from crxml import RenameFields

# Rename "Name" to "name"
stage = RenameFields({"Name": "name"})

What it does¶

For each record, replaces keys according to the mapping. Keys not in the mapping pass through unchanged:

stage = RenameFields({"Name": "name", "Amount": "amount"})

# Input:  {"Name": "Alice", "Amount": 150, "Status": "active"}
# Output: {"name": "Alice", "amount": 150, "Status": "active"}

Fusion¶

Fusable. The Rust engine renames columns during parsing, no Python overhead.

DropFields¶

Removes columns entirely from each record.

from crxml import DropFields

# Drop a single column
stage = DropFields(["InternalId"])

# Drop multiple columns
stage = DropFields(["InternalId", "TempCol"])

What it does¶

For each record, removes keys in the fields set:

stage = DropFields(["InternalId"])

# Input:  {"Name": "Alice", "InternalId": 42, "Amount": 150}
# Output: {"Name": "Alice", "Amount": 150}

Fusion¶

Fusable. The Rust engine skips the dropped column entirely: no scanning, no decoding, no memory allocation for that column.

CastTypes¶

Casts column values to the specified Python types.

from crxml import CastTypes

# Cast "amount" to float
stage = CastTypes({"amount": float})

# Cast multiple columns
stage = CastTypes({"amount": float, "age": int, "active": bool})

What it does¶

For each record, applies the callable to the field value:

stage = CastTypes({"age": int, "amount": float})

# Input:  {"name": "Alice", "age": "30", "amount": "150.5"}
# Output: {"name": "Alice", "age": 30, "amount": 150.5}

If the field is missing from the record, the cast is silently skipped. If the cast fails (e.g., int("abc")), a ValueError is raised:

# This raises ValueError: CastTypes: cannot cast field 'age' value 'abc':
# invalid literal for int()
CastTypes({"age": int})({"age": "abc"})

Fusion¶

Fusable for int, float, bool. The Rust engine parses the column directly as the target type: no string-to-number conversion in Python.

Supported type mappings:

FilterRows¶

Filters rows by a predicate.

from crxml import FilterRows

# Constant filter
stage = FilterRows(field="status", op="==", value="active")

# Column comparison
stage = FilterRows(field_a="price", op=">", field_b="cost")

# Callable predicate
stage = FilterRows(lambda r: r["amount"] > 100)

Constant filter¶

Compares a field to a literal value:

stage = FilterRows(field="status", op="==", value="active")

# Input:  {"name": "Alice", "status": "active"}   → kept
# Input:  {"name": "Bob",   "status": "inactive"} → dropped

Constant filter operators: ==, !=, >, <, >=, <=, starts_with, ends_with

Column comparison¶

Compares two fields in the same record:

stage = FilterRows(field_a="price", op=">", field_b="cost")

# Input:  {"price": 100, "cost": 50}  → kept (100 > 50)
# Input:  {"price": 30,  "cost": 50}  → dropped (30 is not > 50)

Supported operators: >, <, >=, <=, ==, !=, gt, lt, ge, le, eq, ne.

Callable predicate¶

An arbitrary Python function that receives a dict and returns True to keep or False to drop:

stage = FilterRows(lambda r: r["name"].startswith("A"))

Simple lambdas (field comparisons, startswith, compound AND) are automatically compiled into fusable predicates. See Lambda Compiler for the full list of supported patterns and limitations.

Fusion¶

Fusable when using the keyword form (field/op/value or field_a/op/field_b), or when a lambda is automatically compiled by the lambda compiler (simple comparisons, startswith, endswith, arithmetic). The Rust engine applies the filter during parsing. Complex lambdas (closures, nested calls) run in Python.

FilterRowsAny¶

Keeps rows that satisfy any of the given filters (logical OR).

from crxml import FilterRows, FilterRowsAny

stage = FilterRowsAny(
    FilterRows(field="status", op="==", value="active"),
    FilterRows(field="status", op="==", value="pending"),
)

# Keeps rows where status is "active" OR "pending"

Parameters: At least two FilterRows instances (keyword form only).

Fusion¶

Fusable. The Rust engine applies the OR tree during parsing.

FilterRowsAll¶

Keeps rows that satisfy all of the given filters (logical AND).

from crxml import FilterRows, FilterRowsAll

stage = FilterRowsAll(
    FilterRows(field="status", op="==", value="active"),
    FilterRows(field="age", op="!=", value="0"),
)

# Keeps rows where status == "active" AND age != "0"

Parameters: At least two FilterRows instances (keyword form only).

FilterRowsNot¶

Negates a single filter.

from crxml import FilterRows, FilterRowsNot

stage = FilterRowsNot(FilterRows(field="status", op="==", value="deleted"))

# Keeps rows where status != "deleted"

Parameters: Exactly one FilterRows instance (keyword form only).

Combining stages¶

Stages compose freely. The order matters: stages are applied left to right:

from crxml import CrystalXMLSource
from crxml import RenameFields, DropFields, CastTypes, FilterRows
from crxml import FilterRowsAny, FilterRowsNot

src = CrystalXMLSource("report.xml", row_tag="Details")

# Complex pipeline
result = (
    src
    | RenameFields({"Name": "name", "Amount": "amount"})
    | DropFields(["InternalId", "DebugInfo"])
    | CastTypes({"amount": float})
    | FilterRowsAny(
        FilterRows(field="status", op="==", value="active"),
        FilterRows(field="status", op="==", value="pending"),
    )
    | FilterRowsNot(FilterRows(field="name", op="==", value="system"))
)

table = result.to_arrow()

Recap¶

• RenameFields renames columns. Always fusable.
• DropFields removes columns. Always fusable.
• CastTypes converts column types. Fusable for int, float, bool.
• FilterRows filters rows. Fusable when using the keyword form or a compiled lambda.
• FilterRowsAny, FilterRowsAll, FilterRowsNot combine filters.
• Import stages from the adapter package, not from rypipe.

Next: Sinks: materializing pipeline results.