Multi-Output Pipelines

Extract multiple intermediate results from a single optimized execution.

Overview

polars-cv's multi-output feature enables:

• Extracting multiple pipeline stages
• Automatic CSE (Common Subexpression Elimination)
• Single execution pass for all outputs
• Struct column with named fields

Basic Multi-Output

Use .alias() to name checkpoints and dict .sink() for multi-output:

import polars as pl
from polars_cv import Pipeline

# Build pipeline with named checkpoints
base = (
    pl.col("image")
    .cv.pipe(Pipeline().source("image_bytes").resize(height=128, width=128))
    .alias("resized")
)

gray = base.pipe(Pipeline().grayscale()).alias("gray")
thresh = gray.pipe(Pipeline().threshold(128)).alias("thresh")

# Merge branches for multi-output
merged = thresh.merge_pipe(gray)

# Sink multiple outputs
result = df.with_columns(
    outputs=merged.sink({
        "resized": "png",
        "gray": "png",
        "thresh": "png",
    })
)

# outputs is a Struct column with fields: resized, gray, thresh

Extracting Outputs

Access individual outputs from the Struct column:

# Extract individual fields
resized = result.select(pl.col("outputs").struct.field("resized"))
gray = result.select(pl.col("outputs").struct.field("gray"))

# Or extract all at once
extracted = result.select(
    pl.col("outputs").struct.field("resized").alias("resized_img"),
    pl.col("outputs").struct.field("gray").alias("gray_img"),
    pl.col("outputs").struct.field("thresh").alias("thresh_img"),
)

Common Subexpression Elimination (CSE)

Within one .sink(), each distinct value is computed once: the same op on the same inputs is one step, however the pipelines were written. Branches built from separate Pipelines share their decode and whatever leading ops they have in common, branches below a shared node share theirs, and duplicates built twice merge all the way down, operands included. Then each linear run of ops that nothing else reads becomes one node, so a chain written as several .pipe() calls runs as one.

```mermaid flowchart LR subgraph Before["As written"] S1[decode] --> A1[resize] --> B1[grayscale] --> C1[blur] S2[decode] --> A2[resize] --> B2[grayscale] --> D1[threshold] S3[decode] --> F1[flip_h] end

subgraph After["After CSE"]
    S[decode] --> A[resize · grayscale]
    A --> C[blur]
    A --> D[threshold]
    S --> F[flip_h]
end

```

base = Pipeline().source("image_bytes").resize(height=100, width=100)
img = pl.col("image").cv

# Three pipelines, each written as if it stood alone.
blur = img.pipe(base.grayscale().blur(2.0)).alias("blur")
thresh = img.pipe(base.grayscale().threshold(128)).alias("thresh")
flipped = img.pipe(Pipeline().source("image_bytes").flip_h()).alias("flipped")

# One decode, one resize and one grayscale per row feed all three.
result = df.with_columns(
    outputs=blur.merge_pipe(thresh, flipped).sink(
        {"blur": "png", "thresh": "png", "flipped": "png"}
    )
)

expr.sink(..., return_expr=False).show_graph() draws the graph that runs; pass opt_flags=False to sink to see it as written. Every output is byte for byte the same either way.

Across expressions

Polars (2.0 and later) also eliminates common subexpressions between whole plugin calls in one query. The same pipeline written twice — two columns built from one helper, say — runs once:

def thumb() -> pl.Expr:
    return pl.col("image").cv.pipe(
        Pipeline().source("image_bytes").resize(height=64, width=64)
    ).sink("png")

df.lazy().with_columns(a=thumb(), missing=thumb().is_null())  # one decode per row

Polars merges calls whose pipeline and sink are equal. To read several outputs of one pipeline, use a multi-output sink (above), which shares everything but the encode.

Merge Patterns

Merging Multiple Branches

# Create several branches
branch1 = base.pipe(ops1).alias("out1")
branch2 = base.pipe(ops2).alias("out2")
branch3 = base.pipe(ops3).alias("out3")

# Merge all into one graph
merged = branch1.merge_pipe(branch2).merge_pipe(branch3)

Including Earlier Stages

base = expr.cv.pipe(pipe).alias("base")
processed = base.pipe(more_ops).alias("processed")

# Include base in output
merged = processed.merge_pipe(base)
result = df.with_columns(
    outputs=merged.sink({"base": "png", "processed": "png"})
)

Mixed Output Formats

Different outputs can have different formats:

result = df.with_columns(
    outputs=merged.sink({
        "display": "png",           # For display
        "analysis": "numpy",        # For NumPy analysis
        "raw_data": "list",         # As Polars List
    })
)

Per-output sink parameters

Keyword arguments to sink() apply to every output, and a parameter that does not apply to an output's format is refused. To give one output its own parameters, write it as (format, {params}):

result = df.with_columns(
    outputs=merged.sink({
        "x": ("array", {"shape": [224, 224, 3]}),  # array needs its shape
        "thumb": ("jpeg", {"quality": 85}),        # jpeg takes a quality
    })
)

Naming a parameter both in an output's own params and as a shared keyword is refused rather than resolved by precedence.

Multi-Source Multi-Output

Combine multiple input columns with multiple outputs:

# Two different input columns
img = pl.col("image").cv.pipe(Pipeline().source("image_bytes").resize(height=128, width=128))
mask = pl.col("mask_contour").cv.pipe(Pipeline().source("contour").rasterize(shape=img))

# Apply mask
masked = img.apply_mask(mask).alias("masked")

# Merge for multi-output
merged = masked.merge_pipe(img.alias("original")).merge_pipe(mask.alias("mask"))

result = df.with_columns(
    outputs=merged.sink({
        "original": "png",
        "mask": "png",
        "masked": "png",
    })
)

Performance Benefits

Approach	Execution Passes	Memory
Separate .cv.pipe() calls	N (one per output)	N buffers
Multi-output with .sink({...})	1	Shared buffers

Best Practices

1. Use .alias(): Name every checkpoint you want to output
2. Merge All Branches: Call .merge_pipe() to include all in graph
3. CSE is Automatic: No need to manually deduplicate
4. Match Output Names: Sink dict keys must match alias names
5. Extract Early: Pull out needed fields to avoid carrying the whole Struct

Next Steps

• Binary Operations - Combine pipeline outputs