Skip to main content

Node.write()

Store this node in sink (sverify.io.DeltaSink, IcebergSink),

Usage

Source

Node.write(
    sink,
    *,
    quarantine=None,
)

as a cache in the middle of a job or as its output. Each refresh writes it once, upstream first, exactly once per state. Returns the stored node, which transforms read as a versioned source.

Nothing is written until a refresh. Downstream transforms may assume about the stored node what this node’s stage proved: the preconditions on its sources are checked there, not again by the stages that read it. Without a write, transforms fuse into one plan and the intermediate rows are never stored.

Parameters

sink: Sink

Where to store the node, such as DeltaSink(uri) or IcebergSink(catalog, identifier). A sink stores one node.

quarantine: Sink | Literal[False] | None = None
Where the rows this node’s stage sets aside go (on_violation="quarantine", #184): a DeltaSink or IcebergSink whose table holds them, False for no quarantine table (Node.quarantined still returns them), or None (the default): a table next to the sink’s, its URI or identifier with _quarantine appended. Written only when a precondition sets rows aside.

Returns

Node

The stored node, named by the sink’s name. Refreshing it (or

any node that reads it) writes the sink.

Raises

ValueError
sink already stores a node.

Examples

With orders, paid and doubled from the Node example, in a new directory:

lake = tempfile.mkdtemp()
p = orders.pipe(paid).write(DeltaSink(f"{lake}/paid"))
r = p.pipe(doubled).write(DeltaSink(f"{lake}/doubled"))
[w.mode for w in r.refresh().writes]  # ['created', 'created']