Skip to main content

io.DeltaSink

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

Usage

Source

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.

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): 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 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. 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 returns reads the table as a 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). 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.

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) 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. 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.

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). 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:

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() Compact the table’s small files, and with vacuum_retention,
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

Source

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

Source

source()