## Node


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


Usage

``` python
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](Node.pipe.md#sverify.Node.pipe)), or by calling a transform ([sverify.transform](transform.md#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](Node.refresh.md#sverify.Node.refresh) brings the node and every node it reads up to date; the results come from the last refresh, through [to_arrow](Node.to_arrow.md#sverify.Node.to_arrow), [to_polars](Node.to_polars.md#sverify.Node.to_polars), [to_pandas](Node.to_pandas.md#sverify.Node.to_pandas) and [table](Node.table.md#sverify.Node.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](engines.Dialect.md#sverify.engines.Dialect.name). It names the node in [Refresh](nodes.Refresh.md#sverify.nodes.Refresh) and [Status](nodes.Status.md#sverify.nodes.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:

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