---------------------------------------------------------------------- This is the API documentation for the sverify library. ---------------------------------------------------------------------- ## Contracts State what a transformation assumes and promises, and have it proved before any data runs. transform(requires: 'Mapping[str, Sequence[Condition]] | Sequence[Condition] | None' = None, ensures: 'Sequence[Condition]' = (), *, on_unproved: "OnUnproved | Literal['runtime']" = 'error', check_requires: 'bool' = True, check_engine: 'Engine | None' = None, semantics: 'Semantics' = 'reference', on_inferred: 'OnInferred' = 'error', engine: 'Engine | None' = None, infer: 'bool | None' = None, on_violation: 'Policy | None' = None) -> 'Callable[[Callable[..., ir.Table]], ContractedTransform]' Decorate a function from Ibis tables to an Ibis table with a contract. `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. Args: requires: 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: Postconditions on the result, in the same forms (but not `references`). Default: none, which still rules out runtime errors. on_unproved: 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: 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: 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: What the contract's SQL conditions mean in its proofs, as `verify`'s: `"reference"` (the default) or `"duckdb"`. on_inferred: 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: Deprecated alias of `check_engine`. infer: Deprecated: `infer=False` is `on_inferred="off"`. on_violation: 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: 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: ```python 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: ```python 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: ```python 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) ``` verify(query: 'Transform', /, *, requires: 'Mapping[str, Sequence[Condition]] | Sequence[Condition] | None' = None, ensures: 'Sequence[Condition]' = (), tables: 'Mapping[str, ibis.Schema | ir.Table] | None' = None, counterexamples: 'int | None' = None, params: 'Mapping[ir.Scalar, Any] | None' = None, semantics: 'Semantics | None' = None, on_inferred: 'OnInferred' = 'error', infer: 'bool | None' = None) -> 'Report' Try to prove `ensures` about `query` (an Ibis table expression or a SQL query) from `requires`, a mapping from table name to conditions. `tables` gives the input schemas; it is only needed for SQL transforms. For each unproved postcondition, up to `counterexamples` small inputs are tried for one that breaks it (`sverify.counterexample`); 0 skips that. A proof is about the plan `transform` compiles to, under `Rel.eval`. An Ibis expression compiles with sverify's own producer, so it means what `docs/ibis-semantics.md` says, whichever Ibis backend it came from; where Ibis backends disagree (`%` on Polars, `-0.0 == 0.0` on DataFusion, …), no backend's `execute` is guaranteed to match. `sverify.engines.run` runs the verified plan with that meaning (or fails) on DataFusion, DuckDB and Polars. A SQL string, and a SQL condition, is DuckDB's SQL, lifted by `sverify.sql_lift` with `semantics`: the reference semantics by default, or what DuckDB computes with `semantics="duckdb"` (#411). The report says which (`Report.semantics`). Unlike `transform`, `verify` checks nothing on data and raises nothing when a postcondition is unproved: it returns the report. Args: transform: The transformation, as an Ibis table expression or a SQL query. requires: Preconditions, as a mapping from input table name to its conditions (Ibis expressions, SQL snippets, `unique` or `references`), or, for a transformation with one input, a list of its conditions. They are assumed, not checked. Default: none. ensures: Postconditions on the output to prove, including `unique` keys. Default: none, which still reports the facts and possible runtime errors. tables: Input schemas by table name, as Ibis schemas, anything `ibis.schema` accepts, Ibis tables, or mappings of column names to DuckDB's type names (`{"id": "BIGINT", "name": "VARCHAR NOT NULL"}`, `input_schema`). Needed for SQL transforms; an Ibis expression's own tables are found from it. counterexamples: How many small inputs to try per unproved postcondition when searching for one that breaks it; 0 skips the search. Default: the `SVERIFY_COUNTEREXAMPLES` environment variable, else 200 (`default_counterexamples`). params: Values for the `ibis.param`s of an Ibis transform, as Ibis's `execute(params=...)` takes them. The report is about the transform with these values (`bind_params`); an unbound parameter is refused. semantics: What the SQL `verify` lifts means, the transformation if it is a SQL string and every SQL condition: `"reference"` (the default), sverify's reference semantics, where an operation that would create a NaN raises; or `"duckdb"`, what DuckDB computes, NaN and ±inf included. An `sv.sql` expression inside an Ibis transformation keeps its own `semantics`. Unstated (None), it is `"duckdb"` for a transformation whose SQL is all `sv.sql(..., semantics="duckdb")`, else `"reference"`. `"polars"` checks an Ibis transformation under Polars's profile (#444); the checker refuses it until it has one. `"substrait"` checks an Ibis transformation under Substrait's own semantics (`Profile.substrait`, #444): the proof under the reference holds there where Substrait allows nothing more at the options the plan meets, and `Report.transfer` names the options where it allows more. on_inferred: What a precondition sverify infers does. sverify infers the bounds on input columns that rule out integer overflow (`infer_bounds`: `amount * 2` on BIGINT needs `_.amount < 2**62`) and the `IS NOT NULL`s a postcondition needs (`Report.inferred`). `"error"` (the default): the plan is not proved without it, and the report says where it came from and how to state it (`Report.needed`). `"warn"`: it is assumed, checked on the data before each run, and a warning names it (`InferredPreconditionWarning`). `"ignore"`: it is assumed and checked without a warning. `"off"`: nothing is inferred or assumed, so an integer `SUM` that may leave 64 bits, or a double operation that may overflow to ±inf, is an error too, which the other values check on the data while the job runs (`Report.assumed`); each `Issue.hint` suggests a bound. `"silent"` is a deprecated spelling of `"ignore"`. infer: Deprecated: `infer=False` is `on_inferred="off"`. Returns: A `Report` with the status of each postcondition, the possible runtime errors, and what is known about each output column. Raises: ValueError: `requires` names an unknown table, `unique` or `references` names an unknown column or table, or the checker fails on the request (for example on a plan outside the Lean model). Examples: A SQL query (DuckDB's SQL), with its input schema: ```python import ibis import sverify as sv from ibis import _ orders = ibis.table({"id": "int64", "amount": "int64"}, name="orders") report = sv.verify( "select id, amount from orders where amount > 0", tables={"orders": orders}, ensures=["amount > 0"], ) report.proved # True ``` An Ibis expression, with preconditions: ```python report = sv.verify( orders.select(orders.id, total=orders.amount + 1), requires={"orders": [_.amount >= 0]}, ensures=[_.total >= 1], ) report.facts["total"] # 'an integer ≥ 1' ``` unique(*cols: 'str') -> 'Unique' A key: at most one row per combination of `cols`. In `requires` it is an assumption about an input (checked on the data when the input is bound to a backend); in `ensures` it is proved by `Rel.keys_sound` from the input keys and the transformation. Args: *cols: The names of the key's columns. Returns: A `Unique` condition, for `requires` or `ensures`. key(*cols: 'str') -> 'Key' A primary key: at most one row per combination of `cols`, and no NULL in any of them. `sv.key("id")` is `sv.unique("id")` and `'"id" IS NOT NULL'`, which it stands for in the contract and the report. Args: *cols: The names of the key's columns. Returns: A `Key`, for `requires` or `ensures`. references(col: 'str', *, to: 'ir.Column | tuple[str, str] | None' = None) -> 'References' A foreign key: every non-NULL value of `col` occurs in the column `to` of another table. Only for `requires`, where it is checked on the data like `unique` (the referenced table must be bound to the same backend). With a key on the referenced column it lets `Rel.optimize` turn a left join on it into an inner join (`Report.optimized`). Not supported by incremental views. Args: col: The referencing column of the input the condition is on. to: The referenced column, keyword-only: an Ibis column of a named table (`customers.id`), or a `(table, column)` pair of names as `verify` and the backend know them. Returns: A `References` condition, for `requires`. Raises: TypeError: `to` is missing, or is not a column of a named table. Examples: ```python sv.references("customer_id", to=customers.id) sv.references("customer_id", to=("customers", "id")) ``` Report(requires: 'dict[str, list[str]]', ensures: 'list[CheckedCondition]', facts: 'dict[str, str]' = , not_null: 'list[str]' = , deterministic: 'bool' = True, windows_total: 'bool' = True, tie_places: 'list[TiePlace]' = , issues: 'list[Issue]' = , guard_free: 'bool' = False, unchecked: 'dict | None' = None, optimized: 'dict | None' = None, rewrites: 'list[str]' = , inferred: 'dict[str, list[Condition]]' = , implied: 'dict[str, str]' = , assumed: 'list[Issue]' = , params: 'list[str]' = , hints: 'list[str]' = , bounds: 'dict[str, tuple[int | None, int | None]]' = , rows: 'int | None' = None, relations: 'list[RelationRows]' = , semantics: 'Semantics' = 'reference', plan: 'str | None' = None, observation: 'Observation' = , stated: 'bool' = False, floats_on: 'dict[str, list[str]]' = , depends: 'list[dict] | None' = None, routes: 'dict[str, Route]' = , contract: 'dict | None' = None, transfer: 'list[dict] | None' = None, output: 'dict[str, Any]' = , epsilon: 'dict[str, dict[str, Any]] | None' = None, rows_exact: 'dict[str, bool] | None' = None, conditions: 'dict[str, list[Condition]]' = , dialect: "Literal['ibis', 'sql']" = 'ibis', ties: 'dict[str, list[str]]' = , on_inferred: 'OnInferred' = 'ignore', inferred_from: 'dict[str, Issue]' = , postconditions: 'list[Condition]' = , inputs: 'dict[str, ibis.Schema]' = ) -> None What the checker proved about a transformation, returned by `verify` and kept in `ContractedTransform.report`. `str(report)` lists each postcondition with its status, the possible runtime errors and warnings, and what is known about each output column; in a notebook the report renders as tables. from_substrait(data: 'str | bytes | Mapping[str, Any]') -> 'Report' Read a plan `to_substrait` exported, and prove its contract again with the checker (#664). The plan goes to the checker as it was exported, pinned options, shims and all, with the preconditions and postconditions it carries; the report says what the checker proves of it now. For an export of a verified plan, `proved` is true again, with the same postconditions. Args: data: The export: protobuf JSON (a string or its dict) or bytes. Returns: A `Report` of the plan: its postconditions with their status, the runtime errors the checker cannot rule out, the facts on each output column and the options the proof depends on. Facts are in Lean's notation, and nothing is known of the Ibis types. Raises: ValueError: The plan carries no sverify contract, or the checker rejects the plan. ## Report Methods Methods for the Report class to_json(self, indent: 'int | None' = None) -> 'str' The report as a versioned JSON document, for CI jobs and tools (#736), in the format of `sverify/schemas/report.schema.json`: `format` (1), the verdict, each postcondition with its status and `reason`, the runtime errors and assumptions, `needs`, the facts on each output column, the routes engines ran the plan by, and the rest `str(report)` says. Conditions appear as their labels. Within 1.x the format only gains keys (`docs/stability.md`). Args: indent: Indent the JSON by this many spaces; None writes one line. Returns: The JSON document, keys sorted. from_json(data: 'str | Mapping[str, Any]') -> 'Report' A report from what `to_json` wrote, by this version of sverify or an earlier one. Keys this version does not know are left out. The report has the same verdict, postconditions, issues, needs, facts and routes. It holds no plan, and its conditions are their labels, so it does not `run`, `check_inputs` or `to_substrait`. Args: data: The JSON, as text or parsed. Returns: The report. Raises: ValueError: The document's `format` is newer than this version reads. from_substrait(data: 'str | bytes | Mapping[str, Any]') -> 'Report' Read a plan `to_substrait` exported, and prove its contract again with the checker (#664). The plan goes to the checker as it was exported, pinned options, shims and all, with the preconditions and postconditions it carries; the report says what the checker proves of it now. For an export of a verified plan, `proved` is true again, with the same postconditions. Args: data: The export: protobuf JSON (a string or its dict) or bytes. Returns: A `Report` of the plan: its postconditions with their status, the runtime errors the checker cannot rule out, the facts on each output column and the options the proof depends on. Facts are in Lean's notation, and nothing is known of the Ibis types. Raises: ValueError: The plan carries no sverify contract, or the checker rejects the plan. to_substrait(self, format: "Literal['json', 'binary']" = 'json', *, shims: "Literal['apply', 'annotate', 'refuse']" = 'apply') -> 'str | bytes' The verified plan for any conforming Substrait consumer, with its contract as metadata (#664). Every function call names its YAML's URN, its compound name and its output type (Substrait 0.106). Every option the proof depends on (`depends`) is pinned in the call's options at the reference's behaviour. Where Substrait has no option for a choice the proof uses (a spec gap, `upstream/bugs/ST-*.json`), `shims` decides: `"apply"` rewrites the plan into well-specified Substrait, `"annotate"` keeps it and records the choice in the metadata as an assumption the consumer must honour, and `"refuse"` refuses the export. A gap where the reference raises needs no shim in any mode: the preconditions, which travel with the plan, rule those inputs out. The contract travels in the plan's `advanced_extensions` as an `io.sverify.substrait.v1.Contract`: `requires` (declared, inferred and assumed) and `ensures`, each in your terms and as a Substrait plan, and a manifest (the profile, the pinned options, the shims and assumptions, the observation the result holds up to, the Substrait version and extension URNs, sverify's version). `sv.Report.from_substrait` reads it back and proves the contract again. Args: format: `"json"` (protobuf JSON, a string) or `"binary"` (the serialized `substrait.Plan`). shims: What to do at a spec gap the proof depends on. Returns: The plan, as protobuf JSON or bytes. Raises: sverify.errors.ExportError: The report is not proved, or the plan has no conforming Substrait form that keeps the proof; the message names the reason and the gap. substrait_contract(self, *, shims: "Literal['apply', 'annotate', 'refuse']" = 'apply') -> 'dict' The contract `to_substrait` embeds, as JSON, for a sidecar file. to_sql(self, dialect: 'Any' = 'duckdb') -> 'str' The verified plan as SQL whose meaning is the proof's, in `dialect`. The SQL keeps every run-time check, as `run` does. Where the report is `deterministic` (no window ties on data that meets the preconditions), windows need no row-order tiebreak, so Snowflake and Spark take them. A plan verified under an engine's semantics (`semantics="duckdb"`) is written only for an engine its route reaches (`sverify.routes.choose`). Snowflake takes plans verified under the reference. Run the SQL on data that meets the preconditions (`requires`); `run` checks them first (`check_inputs`). Args: dialect: `"duckdb"` (the default), `"snowflake"` or `"spark"`, or one of `sverify.engines.DUCKDB`, `SNOWFLAKE` and `SPARK`. Returns: One SQL query, reading the plan's tables by the names it was verified with. Raises: ValueError: The report has no plan (it was not made by `verify`), or `dialect` is unknown. sverify.errors.UnverifiedRunError: The plan is not proved. sverify.errors.RouteError: No route takes the plan to the dialect's engine. sverify.errors.Unsupported: The dialect cannot express the plan. Examples: ```python report = sv.verify( "SELECT host, max(cpu) AS m FROM t GROUP BY host", tables={"t": t}, ensures=[sv.unique("host")], ) print(report.to_sql("snowflake")) ``` check_inputs(self, con: 'Connection') -> 'None' Check the preconditions (`requires`, the inferred bounds among them) and the `SUM`s assumed to fit (`assumed`) on the tables of `con`, before anything runs: a batch that breaks one is refused with the precondition as the message (#444, goal 1). Raises: PreconditionError: A precondition or an assumed `SUM` does not hold on the data. run(self, con: 'Connection', engine: 'Engine | Any' = None, *, route: "RouteKind | Literal['auto']" = 'auto', unverified: 'bool' = False, deterministic: 'bool' = False, reproducible: 'Observation | None' = None, check_requires: 'bool' = True, streaming: "Literal['allow', 'require']" = 'allow', check_inputs: 'bool | None' = None) -> 'pa.Table' Run the plan this report checked (`plan`) over the tables of `con`, so the result is the one the report is about (#461). The plan reads the tables by the names it was verified with, and runs on `engine` as `sverify.engines.run` runs a plan, with the meaning `Rel.eval` gives it. A plan that is not proved (`proved` is False) is refused (`UnverifiedRunError`), with what was not proved in the message; `unverified=True` runs it anyway, and says so: a warning (`UnverifiedRunWarning`), the route `"unverified"` in `routes`, and `sverify.verified: false` in the result's schema metadata. The run first chooses how the proof reaches the engine (`sverify.routes.choose`, #444): `"auto"` takes the plan natively where it was verified under the engine's profile, through the reference's translation where it was verified under the reference, by the option-wise transfer check, or by a re-check under the engine's profile, and refuses it otherwise. The route goes into `routes`. Where `ensures` names no observation (`sv.equivalent`), the engine's doubles may differ from the reference's within the plan's bound (`epsilon`), except in the columns a postcondition reads. Args: con: The connection whose tables the plan reads (`sv.connect()`). engine: `"duckdb"`, `"polars"`, `"datafusion"`, `"spark"`, or a Spark session. Default: the connection's engine. route: `"auto"` (the default) chooses a route; `"native"`, `"translated"`, `"transfer"` or `"recheck"` takes that one or refuses (`"reference"` is a deprecated alias of `"translated"`). unverified: Run a plan that is not proved, without a route. Default False: such a plan is refused. deterministic: Require a rerun to reproduce the result exactly (`reproducible=sv.equivalent()`, D10). reproducible: What a rerun must reproduce (D20): the engine's run-to-run variation must refine it. Default None: determinism is only reported. check_requires: Check the preconditions (`requires`, the inferred bounds among them) and the assumed `SUM`s on the tables of `con` before the run, and refuse a batch that breaks one (`check_inputs`, #444 goal 1, #604), as `transform(check_requires=)` does on bound inputs. Default True. False skips the check, for a caller that has made it already: the report's guarantee then holds only on data that meets them, and a batch that does not may raise on the engine. streaming: On Polars, `"require"` refuses a plan any of whose nodes runs on the in-memory engine (`polars_engine.fallback_nodes`); `"allow"` (the default) runs it. Either way a chosen route records the nodes in `Route.fallbacks`. check_inputs: Deprecated alias of `check_requires`. Returns: The result as an Arrow table, with the transformation's output names. Raises: ValueError: The report has no plan (it was not made by `verify`), or `engine` is unknown. sverify.errors.UnverifiedRunError: The plan is not proved, and `unverified` is False. It is a `VerificationError` and a `RouteError`. sverify.errors.RouteError: No route takes the plan to `engine`, or, with `streaming="require"`, a node of its Polars plan runs on the in-memory engine. sverify.routes.EngineVersionError: The engine's major version is not the one its profile's facts were probed at. sverify.errors.PreconditionError: With `check_inputs`, a precondition does not hold on the data: nothing ran. Examples: Verify a query, then run what was verified: ```python con = sv.connect() con.put("t", pa.table({"host": ["a", "a", "b"], "cpu": [1, 5, 7]})) report = sv.verify( "SELECT host, max(cpu) AS m FROM t WHERE cpu > 3 GROUP BY host", tables={"t": con.table("t")}, ensures=[sv.unique("host")], ) result = report.run(con) # refused if report.proved is False ``` Report.proved Every postcondition holds and no runtime error is possible. With an observation (`sv.equivalent`), no postcondition depends on what it ignores, and some engine's doubles meet it (`floats_unmet`). Under `semantics="substrait"`, the reference's proof also carries over (`transfer` is empty). Report.floats_unmet Why no engine's doubles meet the observation's `floats` (exact doubles, or a tolerance below the plan's bound), in words; None where one does (`floats_on`). Report.needed With `on_inferred="error"`, the preconditions the proof needs that `requires` does not state, as labels per input table (the bounds `infer_bounds` found, `inferred`); empty otherwise. Report.needed_text `needed` in words: per input table and operation, the preconditions the proof needs, where they come from, and the fixes. Report.equivalence The equivalence the result is proved up to: `sv.equivalent(...)` in `ensures`, or rows as a multiset with exact values (`observation`). Report.needs `needed` as data, one `Needed` per precondition: the table, the column, the comparison and the bound, the SQL, and the runtime error it rules out. A tool that phrases the preconditions in its own terms (a dbt test, an editor's quick fix) reads these; empty unless `on_inferred="error"`. Report.needed_conditions `needs` as conditions, per input table, ready for `requires=`: Ibis deferreds for an Ibis transform (`_.qty < 9 * 10**18`, or `_.qty.isnull() | (_.qty < 9 * 10**18)` where `qty` may be NULL, as the inferred bound lets NULL pass), SQL strings for a SQL one. A bound with no Ibis form goes as its SQL, which an Ibis transform reads too. Empty unless `on_inferred="error"`. report = sv.verify(expr) if not report.proved: report = sv.verify(expr, requires=report.needed_conditions) needs_text(self, table: 'str', labs: 'Sequence[str]') -> 'str' Deprecated: `needed_text` lists every needed precondition. needs_lines(self, needed: 'Mapping[str, Sequence[str]] | None' = None) -> 'list[str]' Deprecated: the property `needed_text`. Report.errors The `issues` an engine fails on at run time (`Issue.error` is true). Report.warnings The `issues` that do not count against a proof. None do: an `SUM` that may leave its type is an error (`errors`). Report.suggestions For each unproved postcondition with one, the preconditions per input table under which it would be proved, as code (`{"(_.y <= 100)": {"orders": ["(_.x <= 50)"]}}`). Report.unproved The postconditions in `ensures` that were not proved. Report.notes What the report notes about the result as a whole, as `str(report)` prints them after "note: ": that rows may tie in an `ORDER BY` (`deterministic` is False, `ties` says which keys would help), and that the proof holds for the `params` given only. route(self, engine: 'Any', route: "RouteKind | Literal['auto']" = 'auto', *, reproducible: 'Observation | None' = None) -> 'Route' The route by which `run` would take the plan to `engine`, and the plan's class there, without running it: the route `report.routes` records after a run (D13). Args: engine: The engine's name (`"duckdb"`, `"polars"`, …) or a Spark session. route: `"auto"` (the default) tries the routes in turn; a route by name (`"native"`, `"reference"`, `"transfer"`, `"recheck"`) is taken or refused. reproducible: What a rerun must reproduce, as `run` takes it. Returns: The `Route`: its `kind`, `cls` (`"exact"`, `"observed"`, `"shim"` or `"refused"`), `observation` and `reason`. A plan that is not proved is refused, with `reason` saying so. Raises: EngineVersionError: The installed engine is another major version than its profile's. EngineSettingsError: A setting the engine's profile depends on differs from its facts'. Report.verdict The report's first line: "proved", or "NOT proved: " and what is missing, counted ("1 of 2 postconditions not proved, 1 possible runtime error"). __str__(self) -> 'str' ## Transforms, nodes and refresh A source piped into a contracted transform, `orders.pipe(paid)`, builds a node. Nodes compose left to right with `.pipe`, are stored with `.write(sink)`, and are kept up to date by `.refresh()`, with update plans proved equal to a full recompute. Node(name: 'str', expr: 'ir.Table', inputs: 'Sequence[Node]') The result of a contracted transform over sources or other nodes. r = orders.pipe(paid).pipe(revenue) # one fused plan r = orders.pipe(paid).write(sink).pipe(revenue) # paid is stored, revenue reads it r.refresh(engine="polars") r.to_pandas() Built by piping a source (`sverify.io`) or a node into a step (`pipe`), or by calling a transform (`sverify.transform`) on it, which is the same: `orders.pipe(paid)` is `paid(orders)`. Its contract is checked then. `write(sink)` stores the node and returns the stored node, which downstream transforms read as a versioned source. `refresh` brings the node and every node it reads up to date; the results come from the last refresh, through `to_arrow`, `to_polars`, `to_pandas` and `table` (an Ibis table). The node keeps its state while the process runs; a stored node's table is its durable state, so a restarted job resumes from it. Attributes: name: The node's name: the transform's, or for a stored node the sink's `name`. It names the node in `Refresh` and `Status`, and a stored node's table in the plans that read it. Examples: Two contracted transforms over an in-memory source, first fused into one plan, then with the first one stored in a Delta table: ```python import tempfile import pyarrow as pa import sverify as sv from ibis import _ from sverify.io import Changes, DeltaSink, StaticSource orders = StaticSource( pa.table({"id": [1, 2], "customer": ["ada", "bob"], "amount": [30, 5]}), name="orders", ) @sv.transform( requires={"orders": [_.amount >= 0, _.amount < 1_000_000_000, sv.unique("id")]}, ensures=[_.amount > 10, _.amount < 1_000_000_000, sv.unique("id")], ) def paid(orders): return orders.filter(orders.amount > 10) @sv.transform( requires={"t": [_.amount > 10, _.amount < 1_000_000_000]}, ensures=[_.doubled > 20], ) def doubled(t): return t.select(t.id, doubled=t.amount * 2) r = orders.pipe(paid).pipe(doubled) # one fused plan, proved here r.refresh() # computed, on DuckDB by default r.to_arrow().to_pylist() # [{'id': 1, 'doubled': 60}] lake = tempfile.mkdtemp() p = orders.pipe(paid).write(DeltaSink(f"{lake}/paid")) # stored: a source r = p.pipe(doubled).write(DeltaSink(f"{lake}/doubled")) r.refresh() # paid first, then doubled; each written once orders.apply(Changes(inserts=[(3, "cy", 50)], deletes=[(1, "ada", 30)])) r.refresh(engine="polars") # updated from the changes alone r.to_arrow().to_pylist() # [{'id': 3, 'doubled': 100}] r.to_polars(); r.to_pandas(); r.table # the same rows, other forms ``` group(*nodes: 'Node') -> 'Group' Several nodes to refresh together, against the same source versions: `sv.group(a, b).refresh()`. A stored node they share is refreshed and written once; `.status()` reports on each. Args: *nodes: The nodes to refresh together: results of transforms, stored or not (not sources). Returns: A `Group`, with `refresh` and `status`. Raises: TypeError: No nodes are given, or one is not a node (a source, or an Ibis table). Examples: Two outputs that read the same stored node, with `p` and `r` from the `Node` example: ```python @sv.transform() def customers(t): return t.select(t.customer) g = sv.group(r, p.pipe(customers)) # fan-out: p is piped twice result = g.refresh() # paid is refreshed and written once [s.node for s in result.nodes] # ['paid', 'doubled', 'customers'] print(g.status()) # one line per node ``` Changes(inserts: 'pa.Table | Sequence[Row]' = , deletes: 'pa.Table | Sequence[Row]' = ) -> None Rows to insert into and delete from one table (bag semantics: each deleted row removes one occurrence). Rows are an Arrow table with the table's columns, or tuples in the table's column order. An update is a delete of the old row plus an insert of the new one. `StaticSource.apply` takes one. Attributes: inserts: Rows to add. Default: none. deletes: Rows to remove, one occurrence each; every one must be in the table, or `StaticSource.apply` raises `ValueError`. Default: none. Group(nodes: 'Sequence[Node]') Nodes refreshed together; see `group`. Attributes: nodes: The nodes, in the order given. Refresh(nodes: 'list[StageRefresh]') -> None What a refresh did, node by node, upstream first: returned by `Node.refresh` and `Group.refresh`. `str()` shows each node's entry; `summary()` one line per node, for a job that refreshes often; `refresh["paid"]` is the entry of the node named "paid". Attributes: nodes: One `StageRefresh` per node refreshed: the nodes asked for and the stored nodes they read, each once, upstream first. Status(nodes: 'list[NodeStatus]' = ) -> None Where each node of a group stands: returned by `Node.status` and `Group.status`. `str()` gives one line per node. Attributes: nodes: One `NodeStatus` per node, in the order they were given. NodeStatus(node: 'str', versions: 'dict[str, int] | None', stored: 'str | None' = None, recomputed: 'str | None' = None, touched: 'str | None' = None, framed: 'bool' = False) -> None Where one node stands; see `Node.status`. `str()` gives one line, such as "doubled: paid@v3; stored in lake/doubled". StageRefresh(node: 'str', report: 'RefreshReport | None', started: 'str | None' = None, write: 'SinkWrite | None' = None, preconditions: 'list[Discharged]' = , retention: 'list[RetentionReport]' = , postconditions: 'list[Discharged]' = , route: 'Route | None' = None, transform: 'str | None' = None, quarantine: 'list[QuarantineReport]' = , quarantined: 'int' = 0, quarantine_table: 'str | None' = None, held: 'str | None' = None, held_error: 'Exception | None' = None) -> None What one refresh did to one node: how its state started (on the first refresh), what the batch was, and what its sink wrote. `str()` shows them in that order; `summary()` gives one line, unless the refresh did something `notable`. ## Node Methods Methods for the Node class __repr__(self) -> 'str' Node.schema The node's columns and their types, known when the node is built. Node.internal_plan The Substrait plan (JSON) of the node's stage, as Lean checked and wrote it: what every engine runs. It uses sverify's own extension URNs, so only sverify reads it (`sverify.engines.to_sql`, `run`, `delta_plans`); `Report.to_substrait()` is the portable plan. `sverify.engines.delta_plans(plan)` are the plans that maintain it from changes. The stage reads the sources and stored nodes upstream under their names, so a stored node's plan is that of the node it stores. Node.plan Deprecated: `internal_plan`, sverify's own form of the plan (`Report.to_substrait()` is the portable one). Node.report The `Report` of the contract that made the node's result: the last transform's proof over its inputs, where a contracted node upstream is a placeholder table under that transform's name, assumed to meet what it guarantees. None for a source, or a node of a transform without a contract. to_substrait(self, format: "Literal['json', 'binary']" = 'json', *, shims: "Literal['apply', 'annotate', 'refuse']" = 'apply') -> 'str | bytes' The node's verified plan for any conforming Substrait consumer, with its contract as metadata: `Report.to_substrait` of `report`. The plan is the last transform's: it reads a contracted node upstream as a table under that transform's name, and the contract's `requires` on it are what that node guarantees. Export each node of a pipeline for the whole of it. `internal_plan` is the fused stage in sverify's internal form. Args: format: `"json"` (protobuf JSON, a string) or `"binary"` (the serialized `substrait.Plan`). shims: What to do at a spec gap the proof depends on (see `Report.to_substrait`). Returns: The plan, as protobuf JSON or bytes. Raises: sverify.errors.ExportError: The node has no contract (a source, or a plain function's node), its report is not proved, or the plan has no conforming Substrait form that keeps the proof. to_sql(self, dialect: 'Any' = 'duckdb') -> 'str' The node's stage as SQL, in `dialect`: what `refresh` runs for a full recompute, with every run-time check, and the meaning the proof gives it. Windows of a `deterministic` report need no row-order tiebreak (`Report.to_sql`). The SQL reads the sources and stored nodes upstream by their names (`internal_plan`). A node whose contract is not proved is refused, as `refresh` refuses it; a node without a contract (a source, or a plain function's result) has no proof to check. Args: dialect: `"duckdb"` (the default), `"snowflake"` or `"spark"`, or one of `sverify.engines.DUCKDB`, `SNOWFLAKE` and `SPARK`. Returns: One SQL query. Raises: ValueError: `dialect` is unknown. sverify.errors.UnverifiedRunError: The node's contract is not proved. sverify.errors.RouteError: No route takes the plan to the dialect's engine. sverify.errors.Unsupported: The dialect cannot express the plan. Examples: ```python node = orders.pipe(paid) print(node.to_sql("snowflake")) ``` pipe(self, step: 'Step', /, **inputs: 'Any') -> 'Node' Apply `step` to this node, left to right: `orders.pipe(paid)` is `paid(orders)`, and a job reads in the order its data flows. The chained node fills the step's first table parameter, and the step's other inputs come by keyword. The contract is checked as when the transform is called: `orders.pipe(paid)` builds the same node, and the same plan, as `paid(orders)`. Args: step: What to apply: a contracted transform (`sverify.transform`); a SQL step, `sverify.sql(query)`, whose inputs are the tables the query reads; or a plain function from Ibis tables to an Ibis table, which is a transform with no contract. **inputs: The step's other inputs, by parameter name: sources, nodes, or plain values. For a SQL step, the keywords are table names in the query, and the chained node is the one table left over. Returns: The node `step` builds. Raises: TypeError: `step` is not a step, or an input is missing, given twice (as the chained node and by keyword) or unknown; the message names the step's parameters. For a SQL step, a table of the query has no input, the chained node is left unused, or `as_` is not the table left over. sverify.errors.CompositionError: A precondition on another node's result does not follow from what that node guarantees. sverify.errors.VerificationError: A postcondition could not be proved. Examples: With `orders`, `paid` and `doubled` from the `Node` example, and a lookup table of customers: ```python customers = StaticSource( pa.table({"customer": ["ada", "cy"], "region": ["eu", "us"]}), name="customers", ) r = orders.pipe(paid).pipe(doubled) # the same node as doubled(paid(orders)) regional = orders.pipe(paid).pipe( # SQL, with the chained node as orders sv.sql( "SELECT o.id, c.region FROM orders o " "JOIN customers c ON o.customer = c.customer" ), customers=customers, ) big = orders.pipe(lambda t: t.filter(t.amount > 20)) # no contract sv.group(r, regional, big).refresh() regional.to_arrow().to_pylist() # [{'id': 3, 'region': 'us'}] ``` write(self, sink: 'Sink', *, quarantine: 'Sink | Literal[False] | None' = None) -> 'Node' Store this node in `sink` (`sverify.io.DeltaSink`, `IcebergSink`), as a cache in the middle of a job or as its output. Each refresh writes it once, upstream first, exactly once per state. Returns the stored node, which transforms read as a versioned source. Nothing is written until a refresh. Downstream transforms may assume about the stored node what this node's stage proved: the preconditions on its sources are checked there, not again by the stages that read it. Without a `write`, transforms fuse into one plan and the intermediate rows are never stored. Args: sink: Where to store the node, such as `DeltaSink(uri)` or `IcebergSink(catalog, identifier)`. A sink stores one node. quarantine: Where the rows this node's stage sets aside go (`on_violation="quarantine"`, #184): a `DeltaSink` or `IcebergSink` whose table holds them, False for no quarantine table (`Node.quarantined` still returns them), or None (the default): a table next to the sink's, its URI or identifier with `_quarantine` appended. Written only when a precondition sets rows aside. Returns: The stored node, named by the sink's `name`. Refreshing it (or any node that reads it) writes the sink. Raises: ValueError: `sink` already stores a node. Examples: With `orders`, `paid` and `doubled` from the `Node` example, in a new directory: ```python lake = tempfile.mkdtemp() p = orders.pipe(paid).write(DeltaSink(f"{lake}/paid")) r = p.pipe(doubled).write(DeltaSink(f"{lake}/doubled")) [w.mode for w in r.refresh().writes] # ['created', 'created'] ``` refresh(self, *, engine: 'Any' = None, audit: 'bool' = False, route: "RouteKind | Literal['auto']" = 'auto', unverified: 'bool' = False, reproducible: 'Observation | None' = None) -> 'Refresh' Bring this node and every node it reads up to date, and write the stored ones. `engine` runs the plans: `"duckdb"` (in memory, the default), `"polars"`, `"datafusion"`, `"spark"` or a Spark session. With `audit`, every batch is compared with recomputation before it is applied (`AuditError`). A stored node's sink commits each state once, by its epoch, so a rerun must reproduce its rows up to the sink's `reproducible` (rows as a multiset, doubles within the plan's bound and other values exact by default): a node whose result is not determined up to row order, or whose engine varies between runs more than that, is refused before anything runs (#444, D10, D20). Each node with a contract is refused unless its report is proved, or `unverified=True` (`Report.run`), or its transform says `on_unproved="warn"`; its proof is taken to `engine` by a route (`sverify.routes.choose`, D13), or the refresh is refused. `StageRefresh.route` records it. Each node is refreshed once per call, upstream first, against each source's latest version when the call starts, and every stored node is written once. The first refresh computes a node in full, or resumes a stored node from its table; later ones apply the changes to the sources with the delta plans Lean derives, or recompute and diff a node that has none (`status` says which). Args: engine: What runs the plans: `"duckdb"`, `"polars"`, `"datafusion"`, `"spark"` or a Spark session. It may differ from one refresh to the next. Default None: DuckDB, in memory, the engine of the nodes' state (as `Report.run`'s None is its connection's). audit: Whether to compare every batch with recomputation before applying it. Default False. route: `"auto"` (the default) chooses a route for each node with a contract; `"native"`, `"translated"`, `"transfer"` or `"recheck"` takes that one or refuses (`"reference"` is a deprecated alias of `"translated"`). unverified: Refresh nodes whose contract is not proved anyway; their route is `"unverified"`, with what was not proved, and an `UnverifiedRunWarning` says so. Default False: such a node refuses the refresh, unless its transform says `on_unproved="warn"`. reproducible: What a rerun must reproduce for every node (`sv.equivalent(...)`, D20), over each sink's own. Default None: a stored node needs its sink's, another node nothing. Returns: A `Refresh`: what was done to each node, upstream first. `print` it for the full report, or log `summary()` for one line per node. Raises: sverify.errors.PreconditionError: A precondition on a source does not hold on its rows (checked in full on the first refresh, then on each batch). sverify.errors.SchemaError: A source's schema changed in a way the contract, proved against the old schema, cannot follow. sverify.errors.AuditError: With `audit`, a batch disagrees with recomputation; nothing was applied. sverify.errors.UnverifiedRunError: A node's contract is not proved (a `VerificationError` and a `RouteError`); nothing was refreshed. sverify.errors.RouteError: A stored node's result is not determined up to row order, or no route takes a node's proof to `engine`; nothing was refreshed. Examples: With `r` from the `Node` example: ```python result = r.refresh(engine="polars") print(result) # one entry per node, upstream first r.to_arrow().to_pylist() # [{'id': 3, 'doubled': 100}] ``` status(self) -> 'Status' Where the node stands: the source versions of its state, where it is stored, and whether it is recomputed rather than maintained. Returns: A `Status` with one `NodeStatus`, for this node. Node.table The node's state as an Ibis table, to query further. Raises: RuntimeError: The node has not been refreshed yet. Node.versions The version of every source the node's state reflects, by name, such as `{"paid": 3}`. A stage reads the sources and stored nodes directly upstream, so these are their versions. Raises: RuntimeError: The node has not been refreshed yet. to_arrow(self) -> 'pa.Table' The node's state as an Arrow table. Returns: The node's rows as of the last refresh. Raises: RuntimeError: The node has not been refreshed yet. quarantined(self, source: 'str | None' = None) -> 'pa.Table' The rows of the node's sources set aside by `on_violation= "quarantine"` (#184), as of the last refresh: each copy once, with the source's columns and `_conditions` (the labels of the preconditions it breaks), `_reason` and `_source_version` (the version that set it aside). Args: source: The source whose rows to return. Default None: the only source with rows set aside, or an empty table. Raises: RuntimeError: The node has not been refreshed yet. ValueError: `source` is None and several sources have rows set aside. to_polars(self) -> 'polars.DataFrame' The node's state as a Polars DataFrame (needs `polars`). Returns: The node's rows as of the last refresh. Raises: RuntimeError: The node has not been refreshed yet. to_pandas(self) -> 'pandas.DataFrame' The node's state as a pandas DataFrame (needs `pandas`). Returns: The node's rows as of the last refresh. Raises: RuntimeError: The node has not been refreshed yet. ## Sources and sinks Delta Lake and Iceberg tables as versioned sources, sources of preconditions, and sinks that carry the contract (`sverify.io`). DeltaSource(uri: 'str', *, storage_options: 'Options' = None, where: 'str | None' = None, name: 'str | None' = None, on_violation: 'Policy | None' = None) A Delta table read by version, as the input of transforms. 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. Args: uri: The table's location: a path, or a URI such as `s3://...` or `uc://catalog.schema.table`. storage_options: 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: 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: 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: 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: ```python 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 ``` DeltaSink(uri: 'str', *, app_id: 'str | None' = None, storage_options: 'Options' = None, name: 'str | None' = None, checkpoint_interval: 'int | None' = None, log_retention: 'Duration | None' = None, compact_every: 'int | None' = None, vacuum_retention: 'Duration | None' = None, reproducible: 'Observation | None' = None) Stores a node in a Delta table, exactly once per state: `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:`. 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`. Args: uri: The sink table's location: a path, or a URI such as `s3://...` or `uc://catalog.schema.table` for a registered table. app_id: The transaction's application id. Default: `"sverify:"`. storage_options: Options for delta-rs's object store, as for `DeltaSource`. Default: None. name: The table's name in the plans that read the stored node. Default: the last part of `uri`, as for `DeltaSource`. checkpoint_interval: Commits between checkpoints (`delta.checkpointInterval`). Default: None, the table's setting (delta-rs: 100). log_retention: 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: Compact the table after every this many writes of this process (`maintain`). Default: None, never. vacuum_retention: 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: 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: 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'} ``` IcebergSource(catalog: 'Catalog', identifier: 'str', *, where: 'str | None' = None, name: 'str | None' = None, on_violation: '_quarantine.Policy | None' = None) An Iceberg table of a catalog, read by snapshot sequence number, as the input of transforms. `name` is the table's name in the plans that read it (default: the last part of `identifier`). A refresh reads the files that appends added since the version a node's state was computed from; after row-level deletes or overwrites it diffs the data files that changed. A table dropped and created again under `identifier` (a new table UUID) is read in full. Tables with equality deletes cannot be read. With `where` (a SQL condition on the columns), the source is the rows on which it is TRUE: snapshots read only the data files whose statistics do not rule it out, and changes are filtered. Args: catalog: A pyiceberg catalog. identifier: The table's identifier in it, e.g. `"db.readings"`. where: A SQL condition on the columns. The source is the rows on which it is TRUE; data files are skipped only where Lean proves from their statistics that it is TRUE on none of their rows (`Rel.eval_pruned` in `SubstraitLean/Pruning.lean`). Default: None, every row. name: The table's name in the plans that read it. Default: the last part of `identifier`. on_violation: 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: catalog: The catalog. identifier: The table's identifier. where: The SQL condition, or None. name: The table's name in the plans that read it. skipped: The data files the last snapshot did not read. IcebergSink(catalog: 'Catalog', identifier: 'str', *, app_id: 'str | None' = None, name: 'str | None' = None, location: 'str | None' = None, partition_by: 'Sequence[PartitionField] | None' = None, properties: 'Mapping[str, str] | None' = None, compact_every: 'int | None' = None, snapshot_retention: 'Duration | None' = None, min_snapshots: 'int' = 1, reproducible: 'Observation | None' = None) Stores a node in an Iceberg table, exactly once per state: `node.write(IcebergSink(catalog, identifier))`, written by every `refresh`. After the first write, each state is one snapshot the size of the change: position delete files for the rows that went (one copy each, so duplicates are exact) and a data file of the rows that came. In a partitioned table, the deletes and the new rows go to the partitions the change touches, and the other partitions keep their files. A new sink on an existing table, or a table that does not hold the rows the sink wrote, is overwritten. Every snapshot carries the contract's plan hash, the epoch (sum of source versions) and the app id; a state whose epoch the last snapshot of this app already has is skipped, unless another plan wrote it: then the table is overwritten, schema included. The contract is also the table property `sverify.contract`, and columns the view proves NOT NULL are `required`. `name` is the table's name in the plans that read the stored node (default: the last part of `identifier`); `app_id` defaults to `sverify:`. The node `write` returns reads the table as an `IcebergSource` 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 holds this state's epoch from other contents of a `StaticSource`. The first write creates the table with `location`, the partition fields of `partition_by` and the table `properties` beside the sink's own, in one commit (#468). A catalog without a default location, such as Unity Catalog's Iceberg REST catalog on a schema without a managed storage root, needs `location`. They only apply to a table the sink creates: an overwrite keeps the table's location, spec and properties, and a sink on an existing table that differs from them warns once (`IcebergSink.create_options_ignored`). Every write adds a snapshot and small files: a refresh that deletes rows adds a position delete file and a data file, which readers open and the sink's next write searches. `compact_every` compacts the table (`maintain`) after every so many writes: in each partition, it rewrites the data files with position deletes, and the small files where there are two or more, as data files of their live rows, and drops the delete files no remaining data file reads, in one `replace` snapshot. With `snapshot_retention`, it then expires the snapshots older than that (keeping the `min_snapshots` newest and the sink's own last snapshot) and deletes the files only they referenced. A compaction snapshot carries the contract, and a restarted job resumes past it, as it does past another writer's compaction. pyiceberg rewrites the table's metadata file on every commit; the table properties `write.metadata.delete-after-commit.enabled` and `write.metadata.previous-versions-max` (`properties`) bound those. Args: catalog: A pyiceberg catalog. identifier: The sink table's identifier; the first write creates it. app_id: The application id recorded in each snapshot. Default: `"sverify:"`. name: The table's name in the plans that read the stored node. Default: the last part of `identifier`. location: The location of the table the first write creates. Default: None, the catalog's default. partition_by: The partition fields of the table the first write creates, each `(column, transform)` or `(column, transform, name)`; a transform is `"identity"`, `"year"`, `"month"`, `"day"`, `"hour"`, `"bucket[N]"`, `"truncate[W]"` or a pyiceberg `Transform`. Default: None, unpartitioned. properties: Table properties of the table the first write creates, such as `"write.target-file-size-bytes"`, besides the sink's own (`sverify.*`). Default: None. compact_every: Compact the table after every this many writes of this process (`maintain`). Default: None, never. snapshot_retention: With `compact_every`, then expire the snapshots older than this and delete the files only they referenced (`maintain(snapshot_retention=...)`). Nodes reading the stored node from an expired snapshot read the whole table, so keep it longer than the gap between their refreshes. Default: None, no expiry. min_snapshots: The newest snapshots of the table's history an expiry keeps, however old. Default: 1. reproducible: 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. Raises: ValueError: A transform of `partition_by` is unknown, or `properties` sets a `sverify.*` property, or `compact_every` or `min_snapshots` is below 1. Attributes: catalog: The catalog. identifier: The sink table's identifier. name: The table's name in the plans that read the stored node. table_location: `location`, or None. partition_by: The partition fields, as `(column, Transform, name or None)`. properties: `properties`, or an empty dict. compact_every: Writes between compactions, or None. snapshot_retention: The expiry's retention after a compaction, or None. min_snapshots: The newest snapshots an expiry keeps. Examples: A sink partitioned by day, at a location of its own: ```python IcebergSink( catalog, "lake.daily", location="s3://lake/daily", partition_by=[("day", "day"), ("host", "identity")], properties={"write.target-file-size-bytes": str(64 << 20)}, ) ``` StaticSource(table: 'pa.Table', *args: 'int', name: 'str | None' = None, version: 'int' = 0, on_violation: 'Policy | None' = None) -> 'None' An in-memory Arrow table as a source, such as a small lookup table or the rows a test feeds in. Its `version` is the caller's (a release number, a batch id), and its `digest` is recorded next to it, so a sink can write a node reading it exactly once. `apply` changes some rows (the next refresh gets exactly that change); `replace` swaps the whole table (the next refresh diffs the two). `name` is the table's name in the plans that read it. Args: table: The table's contents. name: The table's name in the plans that read it. Required: two sources of one name in a pipeline would collide. (Before 1.0 it defaulted to `"table"`, which warns for one release.) version: The version of these contents, by keyword (positional warns for one release). Default: 0. on_violation: 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: table: The current contents. version: Their version. digest: sha256 of their schema and bag of rows. name: The table's name in the plans that read it. SinkWrite(mode: "Literal['created', 'overwrite', 'merge', 'append', 'row_level', 'update', 'skipped']", version: 'int | None', contract: 'Contract', inserted: 'int' = 0, deleted: 'int' = 0, constraints: 'dict[str, str]' = ) -> None What storing a node's state did, in any sink. `mode` is how: `created` the table, `overwrite` it, `merge` the change (Delta), `append` the new rows or write the change `row_level` (Iceberg), `update` it with the change (a sink of your own), or `skipped` a state the table already holds. `version` is the table's version afterwards; `constraints` are the CHECK constraints the table has for the node's guarantees (Delta). `print` it for a one-line summary. A refresh's report lists the writes it made. Attributes: mode: `"created"` (the first write made the table), `"overwrite"` (the table's rows were replaced), `"merge"` (the change in one MERGE commit, Delta), `"append"` (the change only added rows, Iceberg), `"row_level"` (position deletes and new rows in one snapshot, Iceberg), `"update"` (the change, written by a sink of your own) or `"skipped"` (the table already has this state's epoch). version: The table's version afterwards (Delta), its snapshot sequence number (Iceberg), or None if the sink has no versions. contract: What the commit recorded. inserted: Rows written. deleted: Rows deleted. constraints: The table's CHECK constraints from the node's guarantees, by name (Delta; empty for Iceberg). Contract(view: 'str', plan: 'str', sources: 'dict[str, int]', epoch: 'int', not_null: 'list[str]', checks: 'dict[str, str]', unique: 'list[list[str]]', digests: 'dict[str, str]' = , cursors: 'dict[str, Any]' = , table_ids: 'dict[str, str]' = , format: 'int' = 1) -> None What a sink records with each commit: which plan produced the data, from which source versions, and what it proved about every row. `DeltaSource.contract()` and `IcebergSource.contract()` read it back. Attributes: view: The stored node's name. sources: The version of every source table the state was computed from. not_null: Columns the node proves NOT NULL: stated in `ensures` (`_.col.notnull()`), or shown never NULL by its proved report (`Report.not_null`). unique: Column sets the node proves unique. cursors: The cursor of every source that can restore its table from one (`Source.restore`), at its version in `sources`. table_ids: The id of every source table that has one (a Delta table's `metaData.id`), at its version in `sources`. Storage(options: 'Mapping[str, str] | Callable[[], Mapping[str, str]] | None' = None) Options for the object store tables live in (credentials, region, endpoint), given once and passed to every source and sink that needs them: `DeltaSource(uri, storage_options=s3)`. `options` is either a fixed mapping or a function returning a fresh one, called every time a table is opened, so a long-running job can renew an expiring token. sverify keeps no result of that function, and `repr` shows option names, never values. Prefer what the store's default chain finds (environment variables, `~/.aws` profiles, instance or pod roles) or a function that fetches short-lived credentials over literal keys, which travel with any object holding them. The options are delta-rs's (`deltalake`'s `storage_options`), such as `AWS_REGION`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`, `GOOGLE_SERVICE_ACCOUNT` or `AZURE_STORAGE_ACCOUNT_NAME`. A source or sink of your own can take a `Storage` too and call `options()`. Args: options: The options, or a function returning them. Default: none, the store's default chain. Examples: ```python import boto3 from sverify.io import DeltaSink, DeltaSource, Storage def credentials(): c = boto3.Session().get_credentials().get_frozen_credentials() return { "AWS_REGION": "eu-central-1", "AWS_ACCESS_KEY_ID": c.access_key, "AWS_SECRET_ACCESS_KEY": c.secret_key, "AWS_SESSION_TOKEN": c.token or "", } s3 = Storage(credentials) orders = DeltaSource("s3://lake/orders", storage_options=s3) sink = DeltaSink("s3://lake/paid", storage_options=s3) ``` ## Your own sources and sinks Read from an API, a database or a folder of files with one `read` method, and write anywhere with one `write` method; sverify keeps the versions, the changes, the contract and the exactly-once writes. Table formats with metadata implement the full `VersionedSource` protocol. Source(name: 'str', *, schema: 'pa.Schema | None' = None, assumes: 'str | None' = None, on_violation: "Literal['error', 'quarantine', 'hold'] | None" = None) A table you read yourself: subclass it and implement `read`. `read(since)` returns what is new since the cursor of the last read (a `Batch`), or None if nothing is. The first read gets `since=None` and returns the whole table. Each batch that changes something is a new version of the source (1, 2, ...), and a refresh hands the nodes reading it exactly the change between their version and the latest: - `kind="append"` batches are the rows added. That rests on an assumption, such as "rows are only ever appended, in order of `ts`": state it as `assumes`, and the refresh report repeats it. `refresh(audit=True)` reads the whole table again (`reread`) and raises `AuditError` where the assumption broke. - `kind="snapshot"` batches are the whole table. The source diffs each against the last, so the change is exact and nothing is assumed. - `kind="changes"` batches are rows added and removed, trusted as read. The source keeps the table as of its latest version in memory, since joins and keys need it, and has no metadata: preconditions on it are checked on the rows. Versions count reads, and a sink records the version of each source next to its cursor. A restarted job reads the whole table again, unless the source implements `restore`: then it starts from the table as of the cursor its sink recorded, and the first refresh reads only what is new since (`read(cursor)`). A stored node reading a source that cannot restore is recomputed and written again in full, with versions above the ones its sink recorded, so its sink still writes exactly once. Args: name: The table's name in the plans that read it. schema: The table's schema. With it, building a node proves the transform's contract without reading the source, so the first read happens at the first refresh; it is needed when the first read may return nothing. Default: the first batch's. assumes: What the batches rest on, such as `"rows are only appended, in order of ts"`. Default: None. on_violation: 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: name: The table's name in the plans that read it. assumes: What the batches rest on, or None. Examples: A source reading new rows from a list, by an increasing id: ```python import pyarrow as pa import sverify as sv from sverify.io import Batch, Source rows = [{"id": 1, "amount": 30}, {"id": 2, "amount": 5}] class Orders(Source): def read(self, since): new = [r for r in rows if since is None or r["id"] > since] if not new: return None return Batch(pa.Table.from_pylist(new), cursor=new[-1]["id"]) @sv.transform() def big_orders(orders): return orders.filter(orders.amount > 10) big = big_orders(Orders("orders", assumes="ids only grow")) big.refresh() rows.append({"id": 3, "amount": 50}) print(big.refresh()) # big_orders: # orders@v1→v2: 1 insert(s), 0 delete(s); Δ trusted as rows appended # to orders v1→v2, assuming ids only grow # view big_orders: orders@v2 ``` Batch(table: 'Any', cursor: 'Any' = None, kind: 'Kind' = 'append', deletes: 'Any' = None) -> None What a `Source` read: new rows and the cursor to read from next time. Attributes: table: The rows, as an Arrow table, a Polars or pandas DataFrame, a Polars LazyFrame, or anything else `pyarrow.table` accepts. What they are depends on `kind`. cursor: Where the next read starts: `read(since=cursor)`. Any value, such as the greatest timestamp read or the files read so far. Default: None. kind: `"append"`: `table` is rows added since the last read (the default). `"snapshot"`: `table` is the whole table; the source diffs it against the last read. `"changes"`: `table` is rows added and `deletes` rows removed (one copy each). deletes: With `kind="changes"`, the rows removed, in the same forms as `table`. Default: None. from_query(fn: 'Callable[[Any], Any]', *, cursor_on: 'str', name: 'str', schema: 'pa.Schema | None' = None) -> 'Source' A source that reads new rows by a high-water mark: `fn(since)` returns the rows whose `cursor_on` column is greater than `since` (all rows when `since` is None), and the next read starts from the greatest value it returned. This assumes rows only arrive with a greater `cursor_on` than any read before; a late or corrected row is missed. The refresh report says so, and `refresh(audit=True)` catches it: it reads `fn(None)` up to the cursor and compares. Args: fn: Called with the last cursor, or None; returns the new rows (an Arrow table, a Polars or pandas DataFrame, ...) or None. cursor_on: The column whose greatest value is the cursor. name: The table's name in the plans that read it. schema: The table's schema, if `fn(None)` may return nothing. Default: that of the first rows. Returns: The source. Examples: ```python import pyarrow as pa import pyarrow.compute as pc import sverify as sv from sverify.io import from_query payments = pa.table({"id": [1, 2], "ts": [10, 20], "amount": [30, 5]}) def new_payments(since): if since is None: return payments return payments.filter(pc.greater(payments["ts"], since)) @sv.transform() def big(payments): return payments.filter(payments.amount > 10) node = big(from_query(new_payments, cursor_on="ts", name="payments")) node.refresh() ``` from_frame(fn: 'Callable[[], Any]', *, name: 'str', schema: 'pa.Schema | None' = None) -> 'Source' A source that reads the whole table on every refresh: `fn()` returns it. Each read is diffed against the last, so the nodes reading it get the exact change and nothing is assumed; a read that changes nothing is no new version. Args: fn: Returns the table (an Arrow table, a Polars or pandas DataFrame, ...), or None if it is not there. name: The table's name in the plans that read it. schema: The table's schema, if `fn()` may return None. Default: that of the first read. Returns: The source. Examples: ```python import pyarrow as pa from sverify.io import from_frame regions = from_frame( lambda: pa.table({"customer": ["ada"], "region": ["eu"]}), name="regions" ) ``` from_files(path: 'str', *, format: 'str' = 'parquet', name: 'str', schema: 'pa.Schema | None' = None, filesystem: 'Any' = None) -> 'Source' A source that reads the files dropped in a directory: each refresh reads the files that are new since the last, and appends their rows. The cursor is the list of files read. This assumes files are only added: a file changed or removed after it was read is not noticed, except by `refresh(audit=True)`, which reads the files read so far again and compares. Args: path: The directory, as a path or a URI such as `s3://bucket/drop`. format: The files' format and suffix, as `pyarrow.dataset` names it: `"parquet"` (the default), `"csv"`, `"feather"`, `"json"`. name: The table's name in the plans that read it. schema: The table's schema, if the directory may start empty. Default: that of the first files. filesystem: A `pyarrow.fs.FileSystem` to read `path` with, such as one with credentials. Default: the one `path`'s URI names, with its default credentials (environment, profile, role). Returns: The source. Sink(name: 'str', *, app_id: 'str | None' = None, location: 'str | None' = None, reproducible: 'Observation | None' = None) Where `node.write(sink)` stores a node: subclass it and implement `write`. Every refresh hands `write` the node's new state as a `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`), 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` 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` returns a source that reads the sink. A restarted job writes its first state in full. When two processes refresh the same sink, `write` must fail for the one that commits second: raise `sverify.errors.ConcurrentWriteError` if the sink no longer holds what `contract` returned. Its next refresh then writes the whole table, or skips the state the other writer committed. Args: name: The stored table's name in the plans that read it. app_id: Identifies the writer in what the sink records. Default: `"sverify:"`. location: Where the table is, for reports and errors. Default: `name`. reproducible: 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: The stored table's name in the plans that read it. app_id: Identifies the writer. location: Where the table is. reproducible: 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 ``` SinkChange(view: 'IncrementalView', contract: 'Contract', schema: 'pa.Schema', change: 'pa.Table | None') What a sink writes for one state of its node: the change since the state it wrote last, or the whole table. The tables are built when first used, so a sink that reads only `inserts` and `deletes` does not build `table`. Attributes: contract: What to record with the write (`Contract`): the node, the plan, the source versions and what the node proved about every row. `contract.epoch` grows with every state; a sink that stores it can tell the state it holds (`Sink.contract`). schema: The table's schema: the node's columns, NOT NULL where the node proves it. replace: True if the sink's contents must be replaced by `table`: the first write, contents this process did not write, or contents another plan wrote. Then `inserts` is `table` and `deletes` is empty. sink_from(fn: 'Callable[[pa.Table], Any]', *, name: 'str') -> 'Sink' A sink that hands the node's whole table to `fn` whenever it changes: `node.write(sink_from(save, name="paid"))`. Each state is written once per process: a refresh that changes nothing does not call `fn`. A restarted job calls it again with the table it computes. Args: fn: Called with the table (an Arrow table) to store; what it returns is ignored. name: The stored table's name in the plans that read it. Returns: The sink. Examples: ```python import pyarrow as pa import sverify as sv from sverify.io import StaticSource, sink_from orders = StaticSource(pa.table({"id": [1, 2], "amount": [30, 5]}), name="orders") @sv.transform() def big_orders(orders): return orders.filter(orders.amount > 10) big = big_orders(orders).write( sink_from(lambda t: t.to_pandas().to_csv("big.csv"), name="big") ) ``` VersionedSource(*args, **kwargs) What a pipeline needs of a source: a table with integer versions that hands out its snapshot and its change between two versions. Delta and Iceberg tables implement it (`sverify.io`). Sources you write yourself usually subclass `sverify.io.Source` instead, which implements this protocol on top of one `read` method. Implement it directly for a table format with metadata facts and a change feed; `SourceBatch.from_changes` and `from_snapshot` build what `changes` returns. Attributes: name: The table's name in the plans that read it. Snapshot(version: 'int', table: 'pa.Table', facts: 'list[Fact]') -> None A table at one version, as a source hands it to a pipeline: what `VersionedSource.snapshot` returns. Attributes: version: The version. table: The rows at that version. facts: What the metadata says about every row (`sverify.contracts.Fact`), such as file statistics; empty if nothing is known. SourceBatch(start: 'int', end: 'int', snapshot_facts: 'list[Fact]', changes: 'Changes | None' = None, facts: 'list[Fact]' = , trusted: 'str' = '', kind: "Literal['change_feed', 'snapshot_diff']" = 'change_feed', snapshot: 'pa.Table | None' = None, fallback: 'str | None' = None, schema: 'pa.Schema | None' = None) -> None What a versioned source has between a recorded version and its latest one: what `VersionedSource.changes` returns. Either `changes` from the format's change feed (the net change, with `facts` about the inserted rows and the `trusted` assumption it rests on), or the new `snapshot` alone with the reason the feed is unavailable (`fallback`). Build one with `unchanged`, `from_changes` or `from_snapshot`. Attributes: start: The version the batch starts from (`since`). end: The version it reaches. snapshot_facts: What the metadata says about every row at `end`. changes: The change, or None with `snapshot`. facts: What the metadata says about the inserted rows. trusted: What the change rests on, for the refresh report. kind: `"change_feed"` (trusted) or `"snapshot_diff"` (exact). snapshot: The table at `end`, when there is no change. fallback: Why there is no change, with `snapshot`. schema: The table's schema at `end`; None: the one recorded. ## Engines and SQL Run a checked plan on DuckDB, Polars, DataFusion or Spark, or write it as DuckDB, Snowflake or Spark SQL (`Node.to_sql`, `Report.to_sql`; `sverify.engines`). `Dialect`'s SQL-writing methods change as functions land, so they are left out; subclassing it is outside the 1.x promise. connect(engine: 'Engine' = 'duckdb') -> 'Connection' A new, empty connection whose plans run on `engine`. Args: engine: `"duckdb"`, `"polars"` or `"datafusion"` (see `sverify.engines.ENGINES`). Default: `"duckdb"`. Returns: A connection with no tables. Add them with `Connection.create` or `Connection.put`. Raises: ValueError: `engine` is unknown. run_plan(con: 'Connection', plan_json: 'str', engine: 'Engine | Any' = None, *, keep: 'bool' = False, finite: 'bool' = True) -> 'pa.Table | Resident' Run a plan in Lean's written form on `engine` (default: the connection's), over the tables of `con`. DataFusion runs it as `datafusion_plan` writes it; `"spark"` runs it on the local Spark session (`sverify.spark_engine.session`), and a Spark session as `engine` runs it there. DuckDB runs the SQL `sverify.engines.to_sql` writes for the plan, Polars a translation to a lazy frame, Spark the Spark SQL `sverify.engines.to_sql` writes in the `SPARK` dialect; each gives the result `Rel.eval` gives. This is `sverify.engines.run`. Args: con: The connection whose tables the plan reads. plan_json: A plan in Lean's written form (a node's `plan`, or `sverify.engines.written_plan`'s output). engine: `"duckdb"`, `"polars"`, `"datafusion"`, `"spark"`, or a Spark session. Default: the connection's engine. keep: On DuckDB, leave the result in DuckDB, as a `Resident` table. Other engines ignore it. Default: False. finite: On DuckDB and Polars, check that no operation on finite doubles gives `±∞` (`Issue.infinite`, #575), the reference's precondition. Default: True; `Report.run` passes False for a plan verified with `semantics="duckdb"`, which allows `±∞`. Returns: The result as an Arrow table, with the plan's output names; with `keep` on DuckDB, a `Resident` table. Raises: ValueError: `engine` is unknown, or the checker rejects the plan. ImportError: The engine is an extra that is not installed. sverify.engines.Unsupported: The plan uses something the engine's SQL dialect cannot express (DuckDB, Spark). Polars refuses what its translation does not cover with a `ValueError` of its own. Examples: The plan of a transform, on every engine: ```python import ibis import pyarrow as pa import sverify as sv from sverify.engines import ENGINES, run from sverify.io import StaticSource orders = StaticSource(pa.table({"id": [1, 2], "amount": [30, 5]}), name="orders") @sv.transform() def big_orders(orders): return orders.filter(orders.amount > 10) plan = big_orders(orders).internal_plan # the plan the checker proved con = sv.connect() con.create("orders", ibis.schema({"id": "int64", "amount": "int64"}), [(1, 30), (2, 5)]) for engine in ENGINES: assert run(con, plan, engine).to_pylist() == [{"id": 1, "amount": 30}] ``` engines.ENGINES Built-in immutable sequence. If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items. If the argument is a tuple, the return value is the same object. Connection(engine: 'Engine' = 'duckdb') Named Arrow tables, and the engine that runs plans over them. `sverify.connect()` makes one; the class is for type hints. * `create(name, schema, rows)` and `put(name, table)` add a table; `get(name)` is the table as given, the input of every engine, and `rows(name)` its rows. A table may instead be one DuckDB reads where it is: a table format's files (`Scan`), or a table DuckDB holds, such as a view's state (`Resident`). * `table(name)` is an Ibis table bound to it, through an Ibis DuckDB backend that views the Arrow tables. Contracts find a table's data and metadata through it, and Ibis's own `execute` works on it; `con.table(name).get_backend()` is the backend, for SQL of your own. * `engine` runs plans (`sverify.engines.run`) unless a call names another. * The DataFusion engine runs on a private DataFusion session holding the same tables (with the functions Ibis's DataFusion SQL calls), created on first use. Args: engine: `"duckdb"`, `"polars"` or `"datafusion"`. Default: `"duckdb"`. Attributes: engine: The engine that runs plans unless a call names another. arrow: Deprecated: the tables by name, as Arrow tables; `get`, `rows` and `put` read and write them. ibis: Deprecated: the Ibis DuckDB backend the bound tables read through; `table(name).get_backend()` is the same backend. datafusion: Deprecated: the DataFusion session, an internal of the DataFusion engine (#413). Raises: ValueError: `engine` is unknown. Dialect(name: 'str', guards: 'bool' = True, finite: 'bool' = True, decimals: 'bool' = False) -> None SQL syntax and semantics of one engine. Every hook takes SQL text for its operands (each already self-contained, e.g. parenthesised) and returns SQL text. The defaults are ANSI SQL; subclasses override what their engine writes differently and raise `Unsupported` for what it cannot do. Use one of the dialect objects: `sverify.engines.DUCKDB`, `sverify.engines.SNOWFLAKE` and `sverify.engines.SPARK`. This is `sverify.engines.Dialect`. Its fields (`name`, `guards`, `finite`, `decimals`) are public; its SQL-writing methods change whenever a function lands, so they and subclassing are outside the 1.x promise (`docs/stability.md`). Attributes: name: The engine's name, used in `Unsupported` messages. engines.DUCKDB DuckDBDialect(name: 'str', guards: 'bool' = True, finite: 'bool' = True, decimals: bool = True) engines.SNOWFLAKE SnowflakeDialect(name: 'str', guards: 'bool' = True, finite: 'bool' = True, decimals: 'bool' = False) engines.SPARK SparkDialect(name: 'str', guards: 'bool' = True, finite: 'bool' = True, decimals: 'bool' = False) Unsupported Sverify refuses the input: the SQL lift or the Ibis producer cannot read it, Lean's model does not cover it, or an engine or SQL dialect cannot run it. The message says which construct, and why. The SQL lift, the Ibis producer, the SQL writer and the Polars translator raise subclasses of it, so `except sverify.errors.Unsupported` catches every refusal, whichever step made it. It is a `ValueError`. engines.SPARK_CONF dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2) run_sharded(plan: 'str | dict', shards: 'Sequence[Mapping[str, pa.Table]]', *, sharded: 'Sequence[str]', broadcast: 'Mapping[str, pa.Table] | None' = None, keys: 'Mapping[str, Sequence[str]] | None' = None, unique: 'Mapping[str, Sequence[Sequence[str]]] | None' = None, float_order: 'FloatOrder' = 'exact', max_workers: 'int | None' = None, tolerance: 'bool' = False) -> 'pa.Table | Sharded' The plan's result on all the data, computed shard by shard on Polars. This is `sverify.engines.run_sharded`. Each shard maps every table of `sharded` to its rows on that shard; `broadcast` holds the other tables. The result equals the plan's on the concatenated shards, as a bag (row order is not kept), unless `float_order="any"` was needed: sums of doubles then round as the split makes them, within a proved tolerance that `tolerance=True` reports. Before running, the split keys and `unique` column sets are checked on the data. Args: plan: A Substrait JSON plan, as a string or parsed. shards: One mapping per shard, from each table of `sharded` to its rows on that shard. sharded: The names of the tables split across shards. broadcast: The other tables the plan reads, each the same on every shard. Default: none. keys: Split keys, as a mapping from sharded table name to columns: rows with equal values on them are always on the same shard. Default: none. unique: Keys of whole tables, as a mapping from table name to a list of column sets. They let windows run on each shard. Default: none. float_order: `"exact"` (the default) refuses `SUM` and `AVG` of doubles; `"any"` splits them anyway, within a proved tolerance. max_workers: The number of threads running shards in parallel. Default: `ThreadPoolExecutor`'s default. tolerance: Return a `Sharded`: the result, with the tolerance τ of each split double per row (see the module docstring). Default: False. Returns: The plan's result, as an Arrow table, or with `tolerance=True` a `Sharded` holding it. Raises: ValueError: The plan's result would depend on how the rows are split (`sverify.partition.NotShardable`, a `ValueError`, saying why); there are no shards; a shard's tables are not exactly `sharded`; rows with equal split keys are on two shards; a `unique` column set has two equal rows; or `keys` or `unique` name unknown columns. ## ASOF joins Align two time series, each row with the latest earlier row of the other. asof_join(left: ibis.expr.types.relations.Table, right: ibis.expr.types.relations.Table, on: str, by: str | collections.abc.Sequence[str] = (), tolerance: Any = None, *, key: collections.abc.Sequence[str] | None = None, how: Literal['left', 'inner'] = 'left') -> ibis.expr.types.relations.Table Match each left row with the latest right row at or before it. `on` is the time column of both tables; rows match only when their `by` columns are equal and `left.on - tolerance <= right.on <= left.on`. `key` identifies left rows (default: `by` and `on`); the left table must not repeat it. With `how="left"`, left rows without a match are kept with NULLs. The output has the left columns, then the right ones except `by`, with clashing names suffixed `_right` (so `on_right` is the time of the match). Args: left: The table whose rows are matched; one output row per left row. right: The table to take the latest earlier row from. For the result to be determined, `requires` must give it a key on `(*by, on)`. on: The time column, of the same name in both tables. by: Columns that must be equal for rows to match, as one name or a sequence. Default: none. tolerance: How much earlier than the left row a match may be (an Ibis interval, or a number for numeric time columns). Default None: no limit. The join computes `on - tolerance`, which overflows at the type's smallest value; a contract infers the lower bound on the left `on` column that rules it out (#244). key: Columns that identify left rows. Default: `by` and `on`. how: `"left"` (the default) keeps left rows without a match, with NULLs; `"inner"` drops them. Returns: An Ibis table expression, built from a range join and `ROW_NUMBER`. ## Plans The low-level layer: Substrait plans, Lean's written form of them, the SQL of a plan, and the update plans of incremental maintenance (`sverify.engines`). They take sverify's internal plan, so they are outside the 1.x promise and may change in a minor release; `Node.to_sql`, `Report.to_sql` and `Report.to_substrait` are the stable way to a plan. to_sql(plan: 'dict | str', dialect: 'Dialect', *, deterministic: 'bool' = False, guards: 'bool' = True, unchecked: 'dict | None' = None, finite: 'bool' = True, tie_free: 'bool | None' = None) -> 'str' SQL for a Substrait JSON plan, with the semantics `Rel.eval` gives it. The result has the plan's output names (`RelRoot.names`, else `c0`, …), and is ordered as Lean orders it when the plan has a FETCH or a window. Raises `Unsupported` for plans outside the Lean model. This is `sverify.engines.to_sql`. `deterministic=True` promises that no two rows tie on any window's PARTITION BY and ORDER BY columns (what a contract report's `deterministic` says, and what `Report.run(deterministic=True)` requires). Then windows need no ordinal tiebreak, so engines without a stable scan order can run them; FETCH still needs one. `guards=False` leaves out the run-time checks (`Dialect.guards`). Only pass it for a plan whose contract report is `guard_free`, and run the SQL on data that meets the contract's preconditions. `unchecked` leaves them out of some expressions only: pass a contract report's `unchecked` for the same plan, and run the SQL on data that meets its preconditions (`Rel.evalG_marks`). The other expressions, and every aggregate, keep their checks. Args: plan: A Substrait JSON plan, as a string or parsed (`json.loads`), such as a node's `internal_plan` or `sverify.engines.written_plan`'s output. dialect: The engine to write SQL for: `sverify.engines.DUCKDB`, `sverify.engines.SNOWFLAKE` or `sverify.engines.SPARK`. deterministic: Promise that windows have no ties on their PARTITION BY and ORDER BY columns (`Report.deterministic`). Default: False. guards: Write the run-time checks. Default: True. unchecked: The expressions to write without checks, as a contract report's `unchecked` (`sverify.Report.unchecked`) for this plan. Default: None (all of them keep their checks). finite: Write the check that finite doubles do not overflow to `±∞` (`Dialect.finite`), the reference's precondition (#575). Default: True; False for a plan verified with `semantics="duckdb"`. tie_free: Deprecated alias of `deterministic`. Returns: One SQL query in the dialect. It reads the plan's tables by their names, quoted. Raises: Unsupported: The plan is outside the Lean model, its root names do not match its columns, the dialect cannot express it, or `unchecked` is for another plan. Examples: The plan of a transform, as Snowflake SQL: ```python import json import pyarrow as pa import sverify as sv from sverify.engines import SNOWFLAKE, to_sql from sverify.io import StaticSource orders = StaticSource(pa.table({"id": [1, 2], "amount": [30, 5]}), name="orders") @sv.transform() def big_orders(orders): return orders.filter(orders.amount > 10) print(to_sql(big_orders(orders).internal_plan, SNOWFLAKE)) # SELECT s1."c0" AS "id", s1."c1" AS "amount" FROM (SELECT s2."c0" AS "c0", ... ``` substrait_json(con: 'Connection', sql: 'str', semantics: 'str' = 'reference') -> 'str' Compile DuckDB's SQL to a Substrait plan with sverify's lift (`sverify.sql_lift`, SQLGlot), as protobuf JSON. The plan reads the connection's tables by name, and states every choice DuckDB's SQL leaves implicit: casts, output types, function options and sort directions. Args: con: The connection whose tables the SQL reads. sql: A query in DuckDB's SQL dialect. semantics: `"reference"` (the default, as for `sv.verify` and `sv.sql`): what sverify's reference model computes; or `"duckdb"`: what DuckDB computes where the two differ. Before 0.6 the default was `"duckdb"`. Returns: The Substrait plan as protobuf JSON. Raises: Unsupported: The query is outside what the lift reads (a `ValueError`), saying what. written_plan(plan_json: 'str') -> 'str' The plan as Lean parsed it, written back out from its `Rel` (`substrait-check --write`; the golden tests check this round trip), once the checker has found it well-formed. This is the form `sverify.engines.run` and `sverify.engines.to_sql` take, and what a node's `plan` holds. Args: plan_json: A Substrait plan as protobuf JSON. Returns: The plan in Lean's written form, as protobuf JSON. Raises: ValueError: The checker rejects the plan (an operator, function or type outside the Lean model, or an ill-formed plan). run_substrait_json(con: 'Connection', plan_json: 'str') -> 'pa.Table' Run a Substrait plan on DataFusion. The plan runs as given; `sverify.engines.run(con, plan, "datafusion")` first rewrites a plan in Lean's written form where DataFusion's semantics differ, and is what to use for checked plans. Args: con: The connection whose tables the plan reads. plan_json: A Substrait plan as protobuf JSON. Returns: The result, with doubles as `canonical` makes them. Raises: ImportError: DataFusion is not installed. delta_plans(plan_json: 'str') -> 'list[str]' Ask Lean for the weighted delta plans of a Substrait JSON plan; the change to the view is the concatenation of their outputs. Each delta plan reads the view's input tables `t` at their old contents and their changes as tables `t$delta`, which have the columns of `t` and an extra `__weight` column (+1 for an inserted row, -1 for a deleted one). Its output has the view's columns and a weight column likewise. Args: plan_json: A Substrait plan as protobuf JSON, as `sverify.engines.substrait_json` or a node's `plan` write it. Returns: The delta plans, each a Substrait JSON plan in Lean's written form. Raises: sverify.errors.NotIncremental: The plan has an operator without a verified delta rule. ValueError: The checker rejects the plan. ## Testing Test transforms and their contracts with pytest (`sverify.testing`). Installing sverify registers a pytest plugin with the fixtures `sv_source` (in-memory sources) and `sv_engine` (a test per installed engine). assert_contract(node: 'Node', data: 'Mapping[str, Rows] | None' = None, *, expected: 'Rows | None' = None, engine: 'str' = 'duckdb') -> 'pa.Table | None' Assert that a node's contract is proved, and that it holds on `data`. Every contracted transform the node is built from must have its postconditions proved and every precondition on another node's result proved from that node's guarantees. A transform with `on_unproved="error"` (the default) already raises when the node is built; this catches `"warn"` and `"check"`, which let an unproved contract through. With `data`, the node runs on those rows: they must meet the preconditions (a violation is a mistake in the test data, and is reported with the offending rows), and the output must meet the postconditions on `engine`. With `expected`, the output must also be those rows, compared as bags (doubles to 12 significant digits), with the missing and unexpected rows listed when it is not. Args: node: A node: a source or node piped into contracted transforms (`source(...).pipe(paid)`). data: The rows of each table the node reads, by name: its sources, and the stored nodes it reads (`Node.write`). Default: none, which only checks the proof. expected: The rows the node should give on `data`: dicts by column name, tuples in column order, or an Arrow table. engine: Where to run the node: `"duckdb"` (the default), `"polars"`, `"datafusion"` or `"spark"`. Returns: The node's output on `data`, or None without `data`. Raises: AssertionError: The contract is not proved, the data violates a precondition, or the output violates a postcondition or differs from `expected`. ValueError: `data` misses a table the node reads, or has one it does not read. TypeError: `node` is not a node, or `expected` is given without `data`. Examples: ```python import sverify as sv from ibis import _ from sverify.testing import assert_contract, source @sv.transform( requires={"orders": [_.amount >= 0]}, ensures=[_.doubled >= 0], ) def doubled(orders): return orders.select(orders.id, doubled=orders.amount * 2) orders = source([], {"id": "int64", "amount": "int64"}, name="orders") assert_contract(orders.pipe(doubled)) assert_contract( orders.pipe(doubled), data={"orders": [(1, 30), (2, 0)]}, expected=[(1, 60), (2, 0)], ) ``` assert_engines_agree(node: 'Node', data: 'Mapping[str, Rows]', *, engines: 'Sequence[str] | None' = None) -> 'pa.Table' Assert that every engine gives the same rows for a node on `data`. The rows are compared as bags (doubles to 12 significant digits). The data must meet the node's preconditions: outside them the engines may rightly disagree. Each output is also checked against the node's postconditions. Args: node: A node: a source or node piped into contracted transforms. data: The rows of each table the node reads, by name (see `assert_contract`). engines: The engines to compare. Default: `installed_engines()`. Returns: The first engine's output. Raises: AssertionError: The data violates a precondition, an engine fails or violates a postcondition, or two engines give different rows (the message lists the rows that differ from the first engine's). ValueError: Fewer than two engines, or `data` does not match the tables the node reads. Examples: With `doubled` and `orders` from `assert_contract`'s example: ```python from sverify.testing import assert_engines_agree assert_engines_agree(orders.pipe(doubled), {"orders": [(1, 30), (2, 7)]}) ``` cases(node: 'Node', *, max_rows: 'int' = 5) -> 'SearchStrategy[dict[str, pa.Table]]' A Hypothesis strategy of input tables that meet a node's preconditions. Each example maps the name of each table the node reads (as in `assert_contract`'s `data`) to an Arrow table of at most `max_rows` rows. Values are drawn from small ranges, the boundaries of the constants in the node's plans (`sverify.counterexample`) and NULL for nullable columns; then the rows that violate a precondition (evaluated by Lean, `Rel.eval`) are dropped, repeated keys keep their first row, and a foreign key that matches nothing takes a value of the referenced column (or its row goes). So every example meets the preconditions, and the node's guarantees hold on its output. Needs Hypothesis (`pip install 'sverify[testing]'`). Integers, doubles, strings, booleans, dates and timestamps without a time zone are supported. Args: node: A node: a source or node piped into contracted transforms. max_rows: The most rows per table. Default: 5. Returns: A strategy of `{table name: pyarrow.Table}`. Raises: TypeError: A column has a type `cases` cannot generate. ImportError: Hypothesis is not installed. Examples: With `doubled` and `orders` from `assert_contract`'s example: ```python from hypothesis import given, settings from sverify.testing import cases node = orders.pipe(doubled) @settings(max_examples=10, deadline=None) @given(cases(node)) def test_doubled(data): out = assert_contract(node, data=data) assert out.num_rows == data["orders"].num_rows test_doubled() ``` source(rows: 'Rows', schema: 'Any' = None, *, name: 'str | None' = None, version: 'int' = 0) -> 'StaticSource' An in-memory source (`sverify.io.StaticSource`) holding `rows`. The `sv_source` fixture of sverify's pytest plugin is this function. Args: rows: The rows: an Arrow table, dicts by column name, or tuples in the schema's column order. schema: The columns, as anything `ibis.schema` accepts (such as `{"id": "!int64", "amount": "int64"}`, where `!` is NOT NULL), an Ibis schema or an Arrow schema. Optional for an Arrow table. name: The table's name in the plans that read it. Required (before 1.0 it defaulted to `"table"`, which warns for one release). version: The source's version. Default: 0. Returns: A `StaticSource`. Raises: TypeError: No schema is given for rows that are not an Arrow table. installed_engines() -> 'tuple[str, ...]' The engines of `sverify.engines.ENGINES` that can run here: DuckDB always, Polars and DataFusion when their extras are installed. Returns: Engine names, in the order of `ENGINES`. ## Command line The `sverify` command (`sverify.cli`): `check`, `suggest`, `refresh` and `explain` a pipeline's file from a shell, a CI job or an agent. main(argv: 'Sequence[str] | None' = None) -> 'int' Run the `sverify` command with `argv` (default: `sys.argv[1:]`) and return its exit code: 0 proved (a refresh committed every row), 1 not proved, 2 usage error, 3 unsupported or refused, 4 time or memory budget exceeded, 5 a refresh quarantined rows or held a node, 6 error. discover(path: 'str | Path') -> 'Discovery' Import the Python file at `path` without running its pipelines, and prove every contracted transform it applies, as `sverify check` does. A transform applied to sources or nodes is proved on their schemas; one that is not proved, or whose precondition does not follow from the transform feeding it, is recorded instead of raising. A `refresh()` that runs at import raises (put it under `if __name__ == "__main__":`). Args: path: The file. Returns: A `Discovery`: the applications (`checked`), with their reports and the file's variables that hold their nodes; the transforms it never applies (`unchecked`); and what stopped the import, if anything (`error`). Raises: FileNotFoundError: There is no file at `path`. summary(report: 'Report', sources: 'Collection[str] | None' = None) -> 'dict[str, list[str]]' What a report's contract means for the data, in words: * `guarantees`: what every output row (or the output) meets, proved; * `checks`: what each refresh checks on the data before it writes (the preconditions on the sources, postconditions checked on the output, sums assumed to fit). With `sources`, only the preconditions on those tables: the others follow from what the transforms feeding them guarantee; * `not_proved`: what is not proved, and the runtime errors it may raise. Discovery(path: 'Path', checked: 'list[Checked]' = , unchecked: 'list[str]' = , nodes: 'dict[str, Node]' = , transforms: 'dict[str, ContractedTransform]' = , error: 'BaseException | None' = None, module: 'ModuleType | None' = None, _collected: 'list[Collected]' = ) -> None What `discover` found in a file. Attributes: path: The file. checked: Each application of a contracted transform, in the order the file made them. unchecked: Contracted transforms the file defines but never applies: without input schemas they have nothing to be proved on. nodes: The file's nodes, by variable name. transforms: The file's contracted transforms, by variable name. error: What stopped the import, if anything (a refusal, `sverify.errors.Unsupported`; an error of the file's own). module: The imported module, if the import finished. Checked(name: 'str', nodes: 'list[str]', report: 'Report | None', links: 'list[str]', params: 'dict[str, str]', transform: 'ContractedTransform') -> None One application of a contracted transform in the file, and its proof. Attributes: name: The transform's name. nodes: The file's variables that hold the node this application built (empty when no variable does, or it built an Ibis table). report: The `sverify.Report` of its postconditions; None when the budget ran out before it was proved. links: Why a precondition on another transform's result does not follow from that transform's guarantees, one message each. params: For each input table the report names, the parameter of the transform that reads it. transform: The `sverify.ContractedTransform`. ## Standards Contracts in the Open Data Contract Standard (`sverify.odcs`, #791): a node's contract written as ODCS, and ODCS contracts read as conditions. export(node: 'Node', *, id: 'str | None' = None, version: 'str' = '1.0.0', status: 'str' = 'active', description: 'str | None' = None, descriptions: 'Mapping[str, str] | None' = None) -> 'dict[str, Any]' The contract of `node` as an ODCS document (`ODCS_VERSION`). One schema object, named after the node, with a property per column: its logical and physical (DuckDB) type, `required` where the node proves the column NOT NULL, and `primaryKey` or `unique` for its keys. Value sets become `invalidValues` rules, row caps `rowCount` rules, keys over several columns `duplicateValues` rules, and other row conditions `sql` rules (`SELECT COUNT(*) FROM {object} WHERE NOT COALESCE(, FALSE)`, which must be 0). Every flag and rule carries the custom property `sverify` with its `status`: `proved`, `checked` (on the data, `on_unproved="check"`) or `declared`. The document's own `sverify` property names the node, the plan's sha256 (as the sink's contract record has it) and whether the whole contract is proved. Args: node: A node with a contract (a transform's result, stored or not). id: The contract's `id`. Default: a UUID derived from the node's name, the same on every export. version: The contract's `version`. Default "1.0.0". status: The contract's `status`. Default "active". description: The dataset's purpose (`description.purpose` and the schema object's `description`). Default: the first paragraph of the transform's docstring, if any. descriptions: A description per column, by name. Returns: The document as a dict, ready for `json.dump` or `yaml.safe_dump`. Raises: ValueError: `node` has no contract (a source, or a transform without one). write(node: 'Node', path: 'str | Path', **kwargs: 'Any') -> 'dict[str, Any]' Write `export(node, **kwargs)` to `path`: JSON for a `.json` file, YAML otherwise (which needs PyYAML: `pip install sverify[odcs]`). Args: node: The node, as for `export`. path: The file to write. **kwargs: `export`'s keyword arguments. Returns: The document written. read(source: 'str | Path | Mapping[str, Any]', *, strict: 'bool' = False) -> 'OdcsContract' Read an ODCS contract as conditions. Each schema object gives a list of conditions (`OdcsContract.conditions`): `required` columns NOT NULL, the `primaryKey` columns `sv.key`, `unique` columns and `duplicateValues` rules `sv.unique`, value sets (`enum`, `invalidValues` with `validValues`) `IN` lists, the bounds in `logicalTypeOptions` comparisons, `rowCount` caps `sv.max_rows`, foreign keys `sv.references`, and `sql` rules of the form `SELECT COUNT(*) FROM {object} WHERE ` that must be 0 the condition that no row is counted. ODCS leaves NULLs to `required`, so on a column that may be NULL a rule on values lets NULL through. Every other rule (free text, a custom engine's, a pattern, a threshold other than 0, a share of rows) is listed in `unread`, with the reason, and `read` warns. Args: source: A file (YAML, or JSON by its `.json` suffix), the document's text, or the document as a dict. strict: Raise if any rule is unread. Default False: warn. Returns: An `OdcsContract`. Raises: ValueError: The document is no ODCS 3.x contract, or with `strict`, some rule is unread. OdcsContract(document: 'dict[str, Any]', tables: 'dict[str, list[Condition]]' = , unread: 'list[Unread]' = ) -> None An ODCS contract as `read` gives it. Attributes: document: The document as read. tables: The conditions sverify reads, per schema object, in the document's order. unread: The rules sverify cannot read, with the reason. Unread(table: 'str', column: 'str | None', rule: 'Any', reason: 'str') -> None A rule of an ODCS contract that `read` could not read as a condition. Attributes: table: The schema object it is on. column: The property it is on, or None for a rule on the object. rule: The rule (or the property's field) as the document has it. reason: Why sverify cannot read it. UnreadRuleWarning `read` met rules it cannot read as conditions; `OdcsContract.unread` lists them with the reason. ## Errors What sverify raises, and when (`sverify.errors`). SverifyError The base of every exception sverify raises on purpose: a refusal (`Unsupported`), a proof that fails (`VerificationError`), data that breaks a condition (`PreconditionError`), a route, a sink, an export. `except sverify.errors.SverifyError` catches them all; each also keeps the built-in base it had (`ValueError`, `RuntimeError`), so existing handlers still catch it. Misuse of the API (a wrong argument) raises `TypeError` or `ValueError` as usual. VerificationError(report: 'Report') A postcondition, or the absence of runtime errors, could not be proved. Import it from `sverify.errors`. The message's first line counts what is missing; then one line per finding: each needed precondition, each unproved postcondition with what the checker established about it, and each possible runtime error with its hint. `report` holds the rest. Attributes: report: The `Report` of the failed proof. PreconditionError(message: 'str', *, source: 'str | None' = None, condition: 'Condition | None' = None, rows: 'pa.Table | None' = None) Input data violates a precondition. Import it from `sverify.errors`. Raised when a contracted transformation is called on inputs bound to a backend, or a node is refreshed (on the full source tables the first time, then on each batch of changes), and a precondition does not hold on the rows. The message says how many rows violate which condition, for example "2 row(s) of orders violate (_.amount >= 0)". Attributes: source: The input or source table the condition is on (None when the error is about several, an assumed `SUM` among them). condition: The condition as stated in `requires` (an Ibis deferred, a SQL string, a `unique` key, ...) or as inferred, an `Inferred` whose text is its SQL; None as for `source`. rows: Up to `ROWS_SHOWN` (10) of the rows that break it, as an Arrow table with the input's columns; None where the condition is on the table as a whole (`max_rows`, `rows_per`) or the rows could not be fetched. PostconditionError Output data violates a postcondition (only raised by runtime checks). Import it from `sverify.errors`. Raised with `on_unproved="check"`, when a postcondition the checker could not prove fails on the output rows. The message says how many output rows violate which condition. CompositionError(report: 'Report', consumer: 'str', param: 'str', producer: 'str', guarantees) A contracted transformation's precondition on an input is not implied by what the contracted transformation producing that input guarantees. Import it from `sverify.errors`. It is raised when the link is built: when a transform is called on another contracted transformation's result, or on a node (`sverify.Node`) built from one, stored or not. The message names both sides, for example "doubled needs (_.amount >= 0) (it may be FALSE on some rows) on 't'; not_null only guarantees _.amount.notnull()". With `on_unproved="check"` and a bound input, the precondition is checked on the data instead; with `on_unproved="warn"` it is a warning. Attributes: report: The `Report` of the failed proof of the link. consumer: The name of the transformation whose precondition failed. param: The parameter of `consumer` that received the input. producer: The name of the contracted transformation that produced it. NotIncremental The plan is outside the fragment with verified delta rules. Raised by `sverify.engines.delta_plans` for a plan with an operator without a delta rule (a left join, aggregate, window, `DISTINCT` or `LIMIT`). A node over such a plan is recomputed and diffed instead, so refreshing it does not raise this. (A view that ends in `COUNT` and `SUM` per group, or in `DISTINCT`, is maintained from a state per group, `sverify._linear`.) The message is the checker's reason. AuditError In audit mode, the change feed or a view disagrees with recomputation. Raised by `refresh(audit=True)`, which compares each batch with the new snapshots and every node with its recomputation before committing anything: when this is raised, nothing was applied. SchemaError A source's new schema cannot be read in the old one, and a node reading it has a contract that was proved against the old one. Raised by `refresh`. A source whose data fits its old schema (added columns, widened types) is read in the old schema; otherwise the nodes reading it are planned again and recomputed, which a contracted node cannot be. Build its nodes again on the new schema, so the contract is proved against it. Nothing was applied. SqlError A SQL producer refused a query. ExportError The plan has no conforming Substrait form that keeps its proof: the message names the reason, and the spec gap where there is one. Unsupported Sverify refuses the input: the SQL lift or the Ibis producer cannot read it, Lean's model does not cover it, or an engine or SQL dialect cannot run it. The message says which construct, and why. The SQL lift, the Ibis producer, the SQL writer and the Polars translator raise subclasses of it, so `except sverify.errors.Unsupported` catches every refusal, whichever step made it. It is a `ValueError`.