## io.Sink


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


Usage

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


[write](Node.write.md#sverify.Node.write).

Every refresh hands [write](Node.write.md#sverify.Node.write) the node's new state as a [SinkChange](io.SinkChange.md#sverify.io.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](io.Sink.md#sverify.io.Sink.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](io.Sink.md#sverify.io.Sink.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](io.Sink.md#sverify.io.Sink.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](Node.write.md#sverify.Node.write) must fail for the one that commits second: raise `sverify.errors.ConcurrentWriteError` if the sink no longer holds what [contract](io.Sink.md#sverify.io.Sink.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](engines.Dialect.md#sverify.engines.Dialect.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:

``` python
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](#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()](#committed) | Deprecated: [contract()](io.Sink.md#sverify.io.Sink.contract). |
| [contract()](#contract) | The contract of the last state the sink committed, so that a |
| [source()](#source) | The stored table as a versioned source, for the nodes reading the |
| [write()](#write) | Write one state of the node. |

------------------------------------------------------------------------


#### committed()


Deprecated: [contract()](io.Sink.md#sverify.io.Sink.contract).


Usage

``` python
committed()
```


------------------------------------------------------------------------


#### contract()


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


Usage

``` python
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](io.DeltaSink.md#sverify.io.DeltaSink), [IcebergSink](io.IcebergSink.md#sverify.io.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

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

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