io.Sink

Where node.write(sink) stores a node: subclass it and implement

Usage

io.Sink(
    name,
    *,
    app_id=None,
    location=None,
    reproducible=None,
)

write.

Every refresh hands write the node’s new state as a SinkChange: the rows added and removed since the last write, or the whole table, with the contract to record. sverify writes each state exactly once: it skips a state whose contract.epoch the sink has already committed (contract), unless another plan (contract.plan) committed it, or the committed state was computed from other tables (a source with another table id, contract.table_ids, or at a lower version: dropped and created again): then the whole table replaces those rows. By default that record is kept in memory; override contract to read it back from the sink, so a restarted job does not write a state again. The nodes reading the stored node read what this process wrote, with its exact changes, unless source returns a source that reads the sink. A restarted job writes its first state in full.

When two processes refresh the same sink, write must fail for the one that commits second: raise sverify.errors.ConcurrentWriteError if the sink no longer holds what contract returned. Its next refresh then writes the whole table, or skips the state the other writer committed.

Parameters

name: str: The stored table’s name in the plans that read it.
app_id: str | None = None: Identifies the writer in what the sink records. Default: "sverify:<name>".
location: str | None = None: Where the table is, for reports and errors. Default: name.
reproducible: Observation | None = None: What a rerun of a state must reproduce (sv.equivalent(...), #444 D20): the sink commits each epoch once, so a rerun has to give the same rows up to this. Default: rows as a multiset, doubles within the bound the checker proves for the plan (Report.epsilon, sv.equivalent(floats="plan")), other values exact. A sink whose consumers need the same bits on a rerun states sv.equivalent(); one with a fixed tolerance states it, for example sv.equivalent(floats=sv.within(rel=1e-12)).

Attributes

name: str: The stored table’s name in the plans that read it.
app_id: str: Identifies the writer.
location: str: Where the table is.
reproducible: Observation: What a rerun must reproduce.

Examples

A sink that writes each state as a Parquet file, and keeps the contract next to it:

import tempfile
from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq
import sverify as sv
from sverify.io import Contract, Sink, StaticSource


class ParquetSink(Sink):
    def __init__(self, path):
        super().__init__(Path(path).stem, location=str(path))
        self.path = Path(path)
        self.record = self.path.with_suffix(".contract.json")

    def contract(self, version=None):
        if not self.record.exists():
            return None
        return Contract.from_json(self.record.read_text())

    def write(self, change):
        pq.write_table(change.table, self.path)
        self.record.write_text(change.contract.to_json())


@sv.transform()
def big_orders(orders):
    return orders.filter(orders.amount > 10)


orders = StaticSource(pa.table({"id": [1, 2], "amount": [30, 5]}), name="orders")
out = Path(tempfile.mkdtemp()) / "big_orders.parquet"
big = big_orders(orders).write(ParquetSink(out))
print(big.refresh().writes[0])  # big_orders from orders@v0: overwrite, 1 row(s) ...
print(big.refresh().writes[0])  # big_orders from orders@v0: already committed

Attributes

Name	Description
keeps_retired	Whether the sink keeps the rows a view drops from its state

keeps_retired

Whether the sink keeps the rows a view drops from its state

keeps_retired=False

(sverify.arrives_within): it writes each change, and the whole table only when SinkChange.replace. A view with a sink that does not retires nothing.

Methods

Name	Description
committed()	Deprecated: contract().
contract()	The contract of the last state the sink committed, so that a
source()	The stored table as a versioned source, for the nodes reading the
write()	Write one state of the node.

committed()

Deprecated: contract().

Usage

Source

committed()

contract()

The contract of the last state the sink committed, so that a

Usage

Source

contract(version=None)

state is written once. Override it to read the contract back from the sink (see the example).

Parameters

version: int | None = None: The sink’s version to look at, where it keeps a history (DeltaSink, IcebergSink). Default: the latest.

Returns

Contract | None: The contract the last write recorded, or None if there is none.
Default: Contract | None: the one this process wrote last.

Raises

ValueError: A version for a sink that keeps no history.

source()

The stored table as a versioned source, for the nodes reading the

Usage

Source

source()

stored node.

Returns

VersionedSource | None: A source that reads the sink, or None (the default): they read
what this process wrote.

write()

Write one state of the node.

Usage

Source

write(change)

Parameters

change: SinkChange: The change since the last write, or the whole table (change.replace), with the contract to record.

Returns

SinkWrite | None: What the write did, or None: then the refresh report says the
sink wrote change.inserts and change.deletes.