Node

The result of a contracted transform over sources or other nodes.

Usage

Node(
    name,
    expr,
    inputs,
)

r = orders.pipe(paid).pipe(revenue)              # one fused plan
r = orders.pipe(paid).write(sink).pipe(revenue)  # paid is stored, revenue reads it
r.refresh(engine="polars")
r.to_pandas()

Built by piping a source (sverify.io) or a node into a step (pipe), or by calling a transform (sverify.transform) on it, which is the same: orders.pipe(paid) is paid(orders). Its contract is checked then. write(sink) stores the node and returns the stored node, which downstream transforms read as a versioned source. refresh brings the node and every node it reads up to date; the results come from the last refresh, through to_arrow, to_polars, to_pandas and table (an Ibis table).

The node keeps its state while the process runs; a stored node’s table is its durable state, so a restarted job resumes from it.

Attributes

name: The node’s name: the transform’s, or for a stored node the sink’s name. It names the node in Refresh and Status, and a stored node’s table in the plans that read it.

Examples

Two contracted transforms over an in-memory source, first fused into one plan, then with the first one stored in a Delta table:

import tempfile

import pyarrow as pa
import sverify as sv
from ibis import _
from sverify.io import Changes, DeltaSink, StaticSource

orders = StaticSource(
    pa.table({"id": [1, 2], "customer": ["ada", "bob"], "amount": [30, 5]}),
    name="orders",
)

@sv.transform(
    requires={"orders": [_.amount >= 0, _.amount < 1_000_000_000, sv.unique("id")]},
    ensures=[_.amount > 10, _.amount < 1_000_000_000, sv.unique("id")],
)
def paid(orders):
    return orders.filter(orders.amount > 10)

@sv.transform(
    requires={"t": [_.amount > 10, _.amount < 1_000_000_000]},
    ensures=[_.doubled > 20],
)
def doubled(t):
    return t.select(t.id, doubled=t.amount * 2)

r = orders.pipe(paid).pipe(doubled)  # one fused plan, proved here
r.refresh()  # computed, on DuckDB by default
r.to_arrow().to_pylist()  # [{'id': 1, 'doubled': 60}]

lake = tempfile.mkdtemp()
p = orders.pipe(paid).write(DeltaSink(f"{lake}/paid"))  # stored: a source
r = p.pipe(doubled).write(DeltaSink(f"{lake}/doubled"))
r.refresh()  # paid first, then doubled; each written once

orders.apply(Changes(inserts=[(3, "cy", 50)], deletes=[(1, "ada", 30)]))
r.refresh(engine="polars")  # updated from the changes alone
r.to_arrow().to_pylist()  # [{'id': 3, 'doubled': 100}]
r.to_polars(); r.to_pandas(); r.table  # the same rows, other forms