## io.DeltaSink


Stores a node in a Delta table, exactly once per state:


Usage

``` python
io.DeltaSink(
    uri,
    *,
    app_id=None,
    storage_options=None,
    name=None,
    checkpoint_interval=None,
    log_retention=None,
    compact_every=None,
    vacuum_retention=None,
    reproducible=None
)
```


`node.write(DeltaSink(uri))`, written by every [refresh](Node.refresh.md#sverify.Node.refresh).

Each commit carries the transaction `(app_id, epoch)`, where the epoch is the sum of the node's source versions; a state whose epoch the table has already committed is skipped, unless another plan committed it, or it was computed from a source table that has since been dropped and created again ([Contract.table_ids](io.Contract.md#sverify.io.Contract.table_ids)): then the table is overwritten, schema and constraints included. The first write creates the table with Change Data Feed enabled, NOT NULL columns and CHECK constraints for the node's proved guarantees; later writes MERGE the change (each changed row deleted and re-inserted with its new multiplicity) in one commit. A new sink on an existing table first overwrites it, unless it resumes from it. [name](engines.Dialect.md#sverify.engines.Dialect.name) is the table's name in the plans that read the stored node (default: the last part of `uri`); `app_id` defaults to `sverify:<name>`.

A `uri` of the form `uc://catalog.schema.table` writes a table Unity Catalog has registered, with the options described in [DeltaSource](io.DeltaSource.md#sverify.io.DeltaSource). The sink cannot create one: delta-rs writes files and registers no table, so register it in the catalog first, and write its first version at its storage location (`write_deltalake` with the catalog's READ_WRITE keys), or write to its storage location.

The node [write](Node.write.md#sverify.Node.write) returns reads the table as a [DeltaSource](io.DeltaSource.md#sverify.io.DeltaSource), through its Change Data Feed, and builds on the contract the stored node proved. A restarted job that builds the same nodes resumes from the table on its first refresh. The refresh's report includes what each write did ([SinkWrite](io.SinkWrite.md#sverify.io.SinkWrite)). It raises `ValueError` if the table has CHECK constraints the node does not guarantee, or holds this state's epoch from other contents of a [StaticSource](io.StaticSource.md#sverify.io.StaticSource).

Every write adds a commit and small files, which readers replay and open. `checkpoint_interval` and `log_retention` set the table's checkpoint interval and how long its log is kept; `compact_every` compacts the table ([maintain](io.DeltaSink.md#sverify.io.DeltaSink.maintain)) after every so many writes, and `vacuum_retention` then removes the files it replaced. Maintenance commits carry the contract, and a restarted job resumes past them, as it does past another writer's `OPTIMIZE` or `VACUUM`.


## Parameters


`uri: str`  
The sink table's location: a path, or a URI such as `s3://...` or `uc://catalog.schema.table` for a registered table.

`app_id: str | None = None`  
The transaction's application id. Default: `"sverify:<name>"`.

`storage_options: Options = None`  
Options for delta-rs's object store, as for [DeltaSource](io.DeltaSource.md#sverify.io.DeltaSource). Default: None.

`name: str | None = None`  
The table's name in the plans that read the stored node. Default: the last part of `uri`, as for [DeltaSource](io.DeltaSource.md#sverify.io.DeltaSource).

`checkpoint_interval: int | None = None`  
Commits between checkpoints (`delta.checkpointInterval`). Default: None, the table's setting (delta-rs: 100).

`log_retention: Duration | None = None`  
How long the log keeps a commit (`delta.logRetentionDuration`); older commits go when a checkpoint is written. Keep it longer than the gap between two writes of the sink, since the contract is read from the log. Default: None, the table's setting (30 days).

`compact_every: int | None = None`  
Compact the table after every this many writes of this process ([maintain](io.DeltaSink.md#sverify.io.DeltaSink.maintain)). Default: None, never.

`vacuum_retention: Duration | None = None`  
With `compact_every`, then remove the files that no version newer than this needs (`maintain(vacuum_retention=...)`). Nodes reading the stored node that fell further behind read the whole table. Default: None, no vacuum.

`reproducible: Observation | None = None`  
What a rerun of a state must reproduce (`sv.equivalent(...)`, \#444 D20). Default: rows as a multiset, doubles within the plan's bound (`sv.equivalent(floats="plan")`), other values exact; `sv.equivalent()` asks for the same bits.


## Attributes


`uri`  
The sink table's location.

`name: str`  
The table's name in the plans that read the stored node.

`compact_every`  
Writes between compactions, or None.

`vacuum_retention`  
The vacuum's retention after a compaction, or None.


## Examples

A node written to a Delta table and refreshed:

``` python
import tempfile
from pathlib import Path

import pyarrow as pa
import sverify as sv
from deltalake import write_deltalake
from ibis import _
from sverify.io import DeltaSink, DeltaSource

lake = Path(tempfile.mkdtemp())
write_deltalake(lake / "orders", pa.table({"id": [1, 2], "amount": [30, 5]}),
                configuration={"delta.enableChangeDataFeed": "true"})


@sv.transform(ensures=[_.amount > 10])
def big_orders(orders):
    return orders.filter(orders.amount > 10)


orders = DeltaSource(str(lake / "orders"))
big = big_orders(orders).write(DeltaSink(str(lake / "big_orders")))
big.refresh()  # creates the table, with a CHECK constraint for `amount > 10`

write_deltalake(lake / "orders", pa.table({"id": [3], "amount": [50]}), mode="append")
print(big.refresh())  # reads the Change Data Feed, writes one MERGE commit
# big_orders:
#   orders@v0→v1: 1 insert(s), 0 delete(s); Δ trusted as coming from Delta CDF v0→v1
#   view big_orders: orders@v1
#   wrote big_orders from orders@v1: merge v3, 1 row(s) written, 0 deleted
print(DeltaSource(str(lake / "big_orders")).contract().checks)
# {'sverify_1': 'amount > 10'}
```


## Methods

| Name | Description |
|----|----|
| [maintain()](#maintain) | Compact the table's small files, and with `vacuum_retention`, |
| [source()](#source) | The table as a source, for the nodes that read the stored node. |

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


#### maintain()


Compact the table's small files, and with `vacuum_retention`,


Usage

``` python
maintain(*, vacuum_retention=None, vacuum=None)
```


remove the files no version newer than that needs. The commits record the contract of the sink's last commit, if no other writer changed rows since, so a restarted job resumes past them.


##### Parameters


`vacuum_retention: Duration | None = None`  
How old a version may be and still be read, a `timedelta` or an ISO 8601 duration (`"P7D"`), as the constructor's `vacuum_retention`. Nodes reading the stored node from an older version read the whole table. Default: None, no vacuum.

`vacuum: timedelta | None = None`  
Deprecated name of `vacuum_retention`.


##### Returns


`dict[str, object]`  
delta-rs's metrics of the compaction (`"optimize"`) and the

files the vacuum removed (`"vacuumed"`); empty if the table does

not exist.


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


#### source()


The table as a source, for the nodes that read the stored node.


Usage

``` python
source()
```