io.DeltaSource

A Delta table read by version, as the input of transforms.

Usage

io.DeltaSource(
    uri, *, storage_options=None, where=None, name=None, on_violation=None
)

With where (a SQL condition on the columns), the source is the rows on which it is TRUE: snapshots read only the files whose statistics do not rule it out, and changes are filtered. name is the table’s name in the plans that read it (default: the last part of uri).

A uri of the form uc://catalog.schema.table reads the table through Unity Catalog, which delta-rs asks for the table’s location and temporary keys. Its options in storage_options are lowercase (unity_workspace_url, unity_access_token, unity_allow_http_url): delta-rs refuses the uppercase spellings with “Failed to get a credential from UnityCatalog client configuration”. The keys the catalog vends carry no endpoint, so a MinIO or other S3-compatible endpoint goes in the same options (aws_endpoint_url, aws_region, aws_allow_http).

A refresh reads the Change Data Feed (delta.enableChangeDataFeed) between the version a node’s state was computed from and the latest; where the feed is off or its files are gone, the new snapshot is diffed instead. Its metadata gives facts about every row, which can establish a transform’s preconditions without reading the data: NOT NULL columns, CHECK constraints and file statistics.

Parameters

uri: str: The table’s location: a path, or a URI such as s3://... or uc://catalog.schema.table.
storage_options: Options = None: Options for delta-rs’s object store (credentials, region): a Storage shared with other tables, a mapping, or a function returning one, called whenever the table is opened. Default: None, delta-rs’s default chain (environment, profiles, roles).
where: str | None = None: A SQL condition on the columns. The source is the rows on which it is TRUE; files are skipped only where Lean proves from their statistics that it is TRUE on none of their rows. Default: None, every row.
name: str | None = None: The table’s name in the plans that read it. Default: the last part of uri, with dots replaced by underscores (a.b → a_b), or the table’s name for uc://catalog.schema.table.
on_violation: Policy | None = None: What a refresh does with a batch that breaks a precondition on this source, for transforms that set no on_violation of their own: "error", "quarantine" or "hold" (see sverify.transform). Default: None, "error".

Attributes

uri: The table’s location.
name: The table’s name in the plans that read it.
where: The SQL condition, or None.

Examples

A node over a Delta table, updated from its Change Data Feed:

import tempfile
from pathlib import Path

import pyarrow as pa
import sverify as sv
from deltalake import write_deltalake
from sverify.io import DeltaSource

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


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


big = big_orders(DeltaSource(str(lake / "orders")))
big.refresh()
write_deltalake(lake / "orders", pa.table({"id": [3], "amount": [50]}), mode="append")
print(big.refresh())
# big_orders:
#   orders@v0→v1: 1 insert(s), 0 delete(s); Δ trusted as coming from Delta CDF v0→v1
#   view big_orders: orders@v1

Methods

Name	Description
changes()	The change from version `since` to `until`, as a refresh reads it.
contract()	The contract the last sverify commit up to `version` recorded.
latest()	The table’s latest version.
schema()	The table’s schema at its latest version, as Arrow.
snapshot()	The table’s rows at a version, with what its metadata says about
stored()	The node state the table holds, if a DeltaSink stored it: the
table()	The table as delta-rs opens it.

changes()

The change from version since to until, as a refresh reads it.

Usage

Source

changes(since, until=None)

Parameters

since: int: The version the node’s state was computed from.
until: int | None = None: The version to read up to. Default: the latest.

Returns

SourceBatch

The net change from the Change Data Feed, with facts from the

statistics of the files added since; or, where the feed is

unavailable for some commit, the snapshot at until and the

reason.

contract()

The contract the last sverify commit up to version recorded.

Usage

Source

contract(version=None)

Parameters

version: int | None = None: The latest version to look at. Default: the latest.

Returns

Contract | None: The contract a DeltaSink stored in the commit metadata, or None
if no commit has one.

latest()

The table’s latest version.

Usage

Source

latest()

schema()

The table’s schema at its latest version, as Arrow.

Usage

Source

schema()

snapshot()

The table’s rows at a version, with what its metadata says about

Usage

Source

snapshot(version=None)

every row (schema nullability, CHECK constraints, file statistics).

Parameters

version: int | None = None: The version to read. Default: the latest.

Returns

Snapshot: The version, the rows (those passing where, if given) and the
facts.

stored()

The node state the table holds, if a DeltaSink stored it: the

Usage

Source

stored()

last commit is the sink’s, or the commits after the sink’s last one change no rows (OPTIMIZE, VACUUM, table properties and constraints, by another writer or DeltaSink.maintain). A restarted job’s first refresh resumes the sink’s node from it, when it was stored by the same plan from the same sources.

Returns

Stored | None: The stored state, or None if another writer changed rows after
the sink’s last commit.

table()

The table as delta-rs opens it.

Usage

Source

table(version=None)

Parameters

version: int | None = None: The version to load. Default: the latest.

Returns

DeltaTable: A deltalake.DeltaTable.