Transforms and pushdown

Vectorize your transform. Never loop rows in Python. Polars and pyarrow run their ops in Rust over the whole batch. A Python row-loop drops you off that hot path and erases the throughput the pipeline exists to give you.

A pipeline has three stages, and each one keeps the work off Python:

READ (filters=, columns=)  ->  MIDDLE (polars_transform)  ->  WRITE (atomic commit)

• Read prunes rows and columns at the source, before any bytes travel to Python.
• Middle reshapes the Arrow batch with Rust-backed columnar ops.
• Write lands the batch in the sink atomically with its checkpoint.

polars_transform

polars_transform adapts a pl.DataFrame -> pl.DataFrame function into the Arrow-in, Arrow-out callable the engine wants. It is zero-copy both ways over the Arrow C-data interface, so the batch never leaves columnar form. It needs the [polars] extra.

The decorator form reads best on a named function:

import polars as pl
from pyrula.pipelines import polars_transform, postgres_to_delta

@polars_transform
def enrich(df):
    return df.with_columns((pl.col("amount") * pl.col("fx")).alias("usd"))

pipe = postgres_to_delta(
    "postgresql://localhost/prod", "s3://lake/orders",
    table="orders", watermark_column="updated_at",
    pipeline_id="orders-to-delta", transform=enrich,
)
pipe.run()

Inline works too when the transform is a one-liner:

transform=polars_transform(lambda df: df.filter(pl.col("amount") > 0))

You do not have to use Polars. A bare callable that takes an Arrow batch and returns one is still a valid transform, the same contract the quickstart uses with pyarrow.compute. polars_transform is the path when you want Polars expressions without writing the .to_arrow() round-trip yourself. If you forget the round-trip and return a bare pl.DataFrame, the engine normalizes it to pyarrow rather than failing mid-run.

Read pushdown

filters= and columns= narrow the scan at the source. The source reads fewer files and fewer columns, so less data crosses into Python in the first place. The watermark and version checkpoints are untouched, so exactly-once resume still holds.

The filter format follows each source’s native predicate language, so it differs by source.

Postgres takes a raw SQL fragment, AND-ed into the WHERE clause alongside the watermark:

postgres_to_delta(
    "postgresql://localhost/prod", "s3://lake/orders",
    table="orders", watermark_column="updated_at", pipeline_id="orders-us",
    filters="region = 'us' AND amount > 100",   # SQL fragment
    columns=["id", "amount", "fx", "updated_at"],
)

Delta takes a DNF tuple list, the format delta-rs and pyarrow already speak:

delta_to_postgres(
    "s3://lake/orders", "postgresql://localhost/prod",
    table="orders_copy", pipeline_id="delta-to-pg",
    filters=[("region", "=", "us"), ("year", ">", 2023)],   # DNF tuples
    columns=["id", "amount"],
)

filters= is operator config, not user data. Postgres interpolates the fragment into the scan, so never build it from an untrusted caller. There is no honest way to sanitize an arbitrary predicate; the contract is trusted config only.

Schema-typed transforms

Pin a transform’s input or output to a pydantic model, a dataclass, or a pa.Schema. The check is columnar, run once per batch against the Arrow schema, never per row.

import dataclasses

@dataclasses.dataclass
class OrderRow:
    id: int
    amount: float
    usd: float

postgres_to_delta(
    "postgresql://localhost/prod", "s3://lake/orders",
    table="orders", watermark_column="updated_at", pipeline_id="orders-us",
    transform=enrich,
    output_schema=OrderRow,   # checked after the transform
)

input_schema validates the batch before the transform, output_schema after. The engine itself carries no pydantic dependency: the recipe converts the model to a pa.Schema once, and the engine only ever compares Arrow schemas.

slow_path

When a transform genuinely cannot vectorize, set slow_path=True to pass a plain Python callable and silence nothing else. It emits a warning, because per-row Python overhead nullifies the throughput the rest of the pipeline buys you. It exists for one-off debugging, not production.

postgres_to_delta(..., transform=row_by_row_thing, slow_path=True)

If the logic is heavy enough to need this, it usually belongs in a workflow step instead, off the pipeline hot path.