Skip to main content

transform()

Decorate a function from Ibis tables to an Ibis table with a contract.

Usage

Source

transform(
    requires=None,
    ensures=(),
    *,
    on_unproved="error",
    check_requires=True,
    check_engine=None,
    semantics="reference",
    on_inferred="error",
    engine=None,
    infer=None,
    on_violation=None
)

requires maps parameter names to conditions on that input. On each call the postconditions are proved (cached per input schema and expression) and the report is kept in .report; see the module docstring. With on_unproved="check", unproved postconditions are checked on the output data instead: on inputs bound to a connection, when the transform is called; on a node, at every refresh, on the rows the batch inserts, before its sink commits. The report marks them checked. check_engine says where those checks on data run (see the module docstring).

Piped from a source (sverify.io) or a node, orders.pipe(doubled), or called on one, doubled(orders), the transform builds a sverify.Node instead of an Ibis expression: transforms compose over sources, .write(sink) stores a result, and refresh keeps it up to date (see sverify.Node). Node.pipe also takes a plain function, which is a transform with no contract.

Parameters

requires: Mapping[str, Sequence[Condition]] | Sequence[Condition] | None = None

Preconditions, as a mapping from parameter name to conditions on that input (or, for a function of one table, a list of its conditions): Ibis expressions (_.amount >= 0), functions of the table, SQL snippets ("amount >= 0"), unique or references. Default: none.

ensures: Sequence[Condition] = ()

Postconditions on the result, in the same forms (but not references). Default: none, which still rules out runtime errors.

on_unproved: OnUnproved | Literal["runtime"] = "error"

What happens when a postcondition, or a precondition on another contracted transformation’s result, is not proved: "error" (the default) raises; "check" checks a postcondition on the output data instead (bound inputs, or a node at every refresh) and a precondition on a bound input’s data; "warn" emits a warning and returns the result, and a node’s refreshes run it unverified, each with an UnverifiedRunWarning (refresh(unverified=True)). "runtime" is a deprecated spelling of "check".

check_requires: bool = True

Whether to check preconditions on the data of bound inputs at call time. Default True. False with an on_violation other than "error" raises ValueError: that setting acts on those checks.

check_engine: Engine | None = None

Where checks on data run: "duckdb", "polars" or "datafusion" run each condition’s plan as Lean wrote it; None (the default) runs them through Ibis on the inputs’ connection. refresh(engine=) says where a node’s plan runs.

semantics: Semantics = "reference"

What the contract’s SQL conditions mean in its proofs, as verify’s: "reference" (the default) or "duckdb".

on_inferred: OnInferred = "error"

What a precondition sverify infers does, as verify’s. sverify infers the bounds it needs (amount * 2 on BIGINT needs _.amount < 2**62). "error" (the default): the contract is not proved until requires states it or the transformation no longer needs it (Report.needed). "warn": it is assumed and checked on the data or proved from table metadata, with an InferredPreconditionWarning. "ignore": the same without the warning. "off": nothing is inferred or assumed, so an integer SUM that may leave 64 bits is an error too, which the other values check on the data while the job runs (Report.assumed). "silent" is a deprecated spelling of "ignore".

engine: Engine | None = None

Deprecated alias of check_engine.

infer: bool | None = None

Deprecated: infer=False is on_inferred="off".

on_violation: Policy | None = None
What a refresh does with a batch of a source that breaks one of these preconditions, inferred ones included (#184): "error" raises PreconditionError; "quarantine" sets the offending rows aside with the reason, in the node’s quarantine table, and the rest flows; "hold" commits nothing for the batch and holds the node at its versions until a later version of the source meets the preconditions. Default None: the source’s on_violation, else "error". Where two transforms state the same condition, the stricter holds.

Returns

Callable[[Callable[…, ir.Table]], ContractedTransform]
A decorator that turns the function into a ContractedTransform.

Raises

TypeError
When decorating, if requires names a parameter the function does not have. Calls raise PreconditionError, CompositionError, VerificationError or PostconditionError from sverify.errors (see ContractedTransform.__call__).

Examples

A contract that holds, on a source, and what the checker worked out about the output:

import pyarrow as pa
import sverify as sv
from ibis import _
from sverify.io import StaticSource

orders = StaticSource(
    pa.table({"id": [1, 2], "amount": [30, 5]}), name="orders"
)

@sv.transform(
    requires={"orders": [_.amount >= 0, _.id.notnull()]},
    ensures=[_.doubled >= 0, _.doubled.notnull(), "id IS NOT NULL"],
)
def doubled(orders):
    return orders.filter(orders.amount > 10).select(orders.id, doubled=orders.amount * 2)

big = orders.pipe(doubled)  # a Node, proved here (as doubled(orders))
doubled.report.proved  # True
doubled.report.facts["doubled"]  # 'an integer ≥ 22'
big.refresh()
big.to_arrow().to_pylist()  # [{'id': 1, 'doubled': 60}]

Transforms compose: the downstream precondition is proved from the upstream guarantee, and a link that does not hold is refused:

from sverify.errors import CompositionError

@sv.transform(requires={"t": [_.doubled >= 0]}, ensures=[_.plus >= 1])
def plus_one(t):
    return t.select(t.id, plus=t.doubled + 1)

big.pipe(plus_one)  # proved from what doubled guarantees

@sv.transform(requires={"t": [_.doubled >= 100]})
def huge(t):
    return t

try:
    big.pipe(huge)
except CompositionError as e:
    print(str(e).split(";")[0])
# huge needs (_.doubled >= 100) (it may be FALSE on some rows) on 't'

A postcondition that does not follow is refused:

from sverify.errors import VerificationError

@sv.transform(requires={"orders": [_.amount >= 0]}, ensures=[_.total > 5])
def total(orders):
    return orders.select(orders.id, total=orders.amount + 1)

try:
    orders.pipe(total)
except VerificationError as e:
    print(str(e).splitlines()[0])
# could not prove: (_.total > 5) (it may be FALSE on some rows)