Skills
A skill is a package of structured files that teaches an AI coding agent how to work with a specific tool or framework. This project ships multiple skills — use the switcher below to browse each one. Install a skill in your agent and it will be able to run commands, edit configuration, write content, and troubleshoot problems without step-by-step guidance from you.
Any agent — install all with npx:
npx skills add https://alexandernenninger.github.io/substrait-lean/CLI — install all skills in a project:
great-docs skill install sverifyCodex / OpenCode
Tell the agent to fetch these skill files:
https://alexandernenninger.github.io/substrait-lean/.well-known/agent-skills/sverify/SKILL.md
https://alexandernenninger.github.io/substrait-lean/.well-known/agent-skills/sverify-pipelines/SKILL.md
https://alexandernenninger.github.io/substrait-lean/.well-known/agent-skills/sverify-engines/SKILL.mdOr browse the skill files below.
SKILL LAYOUT
sverify/
├── SKILL.md
└── references/
└── contracts.md
SKILL.md
--- name: sverify description: > Write data transformations with proved contracts using the sverify Python package. Use when a user wants to state what an Ibis or SQL transformation assumes about its inputs and guarantees about its output (not NULL, unique keys, value ranges, no division by zero), check those guarantees before the pipeline runs, read a sverify Report, fix a postcondition sverify could not prove, or test transforms and their contracts with pytest. --- # sverify: contracts proved before a pipeline runs `sverify` hands a transformation's query plan to a checker written in Lean, which proves the postconditions (`ensures`) for every input that meets the preconditions (`requires`), or says why it can't. Use it in place of data tests on derived data: the checks hold on all inputs, not only on the data that happened to be there. ## Install ```bash pip install sverify # core: DuckDB, contracts, views pip install 'sverify[datafusion]' # DataFusion as an engine (SQL needs no extra) pip install 'sverify[all]' # every engine and table format ``` Wheels bundle the checker, so no Lean toolchain is needed. ## Write a contracted transformation The function body is ordinary Ibis. Called on a source, it proves the contract and returns a `Node`; `refresh()` computes it, on DuckDB unless you pass `refresh(engine=...)`. `report` says what was proved. `@sv.transform(check_engine=...)` says where checks on data run. ```python import ibis import pyarrow as pa from ibis import _ import sverify as sv from sverify.io import StaticSource orders = StaticSource( # in memory; DeltaSource and IcebergSource work the same way 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=[_.doubled >= 22, _.doubled.notnull(), sv.unique("id")], ) def big_orders(orders): return orders.filter(orders.amount > 10).select(orders.id, doubled=orders.amount * 2) big = big_orders(orders) # proved here, before any data runs big.refresh() assert big.to_arrow().to_pylist() == [{"id": 1, "doubled": 60}] report = big_orders.report assert report.proved print(report.facts["doubled"]) # what Lean established about the column ``` * Conditions are Ibis expressions over the output (`_.doubled >= 22`) or SQL snippets (`"doubled >= 22"`). `sv.unique(...)` states a key; `sv.references("cust", to=customers.id)` a foreign key (only in `requires`); `sv.max_rows(n)` caps an input's rows. A transformation with one input also takes a plain list, `requires=[_.amount >= 0]`. * `requires` is checked on the source's data when the node is refreshed (`sverify.errors.PreconditionError`), or proved from table metadata. * Transforms compose: `revenue(paid(orders))` proves each link from the upstream contract (`sverify.errors.CompositionError` names a broken one). * Called on Ibis tables instead of a source, a transform returns an Ibis expression: handy for checking a contract without data. ## Read the report, and fix what isn't proved An unproved postcondition raises `sverify.errors.VerificationError`. `sv.verify` returns the report without raising, with a counterexample when the search finds one: ```python orders_t = ibis.table({"id": "int64", "customer": "string", "amount": "int64"}, name="orders") report = sv.verify( orders_t.select(orders_t.id, total=orders_t.amount + 1), requires={"orders": [_.amount >= 0]}, ensures=[_.total > 5], ) failed = report.unproved[0] print(failed.label, failed.value) # (_.total > 5) TRUE or FALSE print(failed.counterexample) # an input row with amount = 0 gives total = 1 ``` Read the counterexample, then either strengthen `requires` (if the input really can't look like that), change the transformation, or weaken the postcondition. Here a filter makes it hold: ```python fixed = orders_t.filter(orders_t.amount > 4).select(orders_t.id, total=orders_t.amount + 1) bounded = {"orders": [_.amount >= 0, _.amount < 1_000_000_000]} assert sv.verify(fixed, requires=bounded, ensures=[_.total > 5]).proved ``` `report.issues` lists runtime errors that may happen (division by zero, overflow at the declared width, bad casts); an error makes `proved` False. Sverify infers the bounds that rule most of them out (`amount * 2` needs `_.amount < 2**62`) and, by default (`on_inferred="error"`), reports each under `needs:` (`report.needed`) until `requires` states it or the transformation no longer needs it (filter those rows first, or a CASE); `on_inferred="warn"` or `"ignore"` assumes it and checks it on the data, and `on_inferred="off"` infers and assumes nothing. For the rest, `report.hints` names the bound to add, and an unproved comparison with a constant comes with the precondition that proves it (`report.suggestions`). An integer `SUM` that may overflow is listed in `report.assumed` and checked at run time (an error with `on_inferred="off"`); a row cap (`sv.max_rows(n)`) proves it fits. Guard a divisor with `NULLIF`/`ifelse`, or bound it in `requires`. Sums and means of doubles round: with a row cap, `sv.approx(_.cpu_avg <= 100)` states a postcondition up to the proved rounding error. ## Choose what happens when a proof fails `@sv.transform(..., on_unproved=...)`: * `"error"` (default): raise `VerificationError`. * `"check"`: check the unproved postconditions on the output data instead, at refresh (raises `PostconditionError`). * `"warn"`: warn and continue; a node's refreshes run it unverified, each with an `UnverifiedRunWarning`. A run or refresh of a plan that is not proved raises `UnverifiedRunError`, a `VerificationError` (and a `RouteError`), unless you pass `unverified=True`. ## Test transforms with pytest Installing sverify registers a pytest plugin. In a test suite, assert the contract, run the node on a few rows, and compare the engines; with `expected=`, a failure lists the missing and unexpected rows, and test data that breaks a precondition is reported with its rows. ```python from sverify.testing import assert_contract, assert_engines_agree, cases, source def test_big_orders(): # with the plugin: def test_big_orders(sv_source, sv_engine) orders = source([], {"id": "int64", "customer": "string", "amount": "int64"}, name="orders") node = orders.pipe(big_orders) assert_contract(node) # every contract in the node's lineage is proved data = {"orders": [(1, "ada", 30), (2, "bob", 5)]} out = assert_contract(node, data=data, expected=[{"id": 1, "doubled": 60}]) assert out.num_rows == 1 assert_engines_agree(node, data) # every installed engine gives the same rows test_big_orders() ``` * `data` names every table the node reads: its sources, and the stored nodes (`.write`) it reads. Rows are tuples in column order, dicts, or Arrow tables; `expected` is compared as a bag. * Fixtures: `sv_source(rows, schema, name=...)` is `source`; a test taking `sv_engine` runs once per installed engine (pass it as `engine=`); `pytest --sv-engines=duckdb,polars` picks them, `-p no:sverify` turns the plugin off. * `cases(node, max_rows=5)` is a Hypothesis strategy of input tables that meet the preconditions (`pip install 'sverify[testing]'`), for property tests of the code around a node: ```python from hypothesis import given, settings ORDERS = {"id": "int64", "customer": "string", "amount": "int64"} node = source([], ORDERS, name="orders").pipe(big_orders) @settings(max_examples=10, deadline=None) @given(cases(node)) def test_doubles_are_big(data): out = assert_contract(node, data=data) assert all(d > 20 for d in out["doubled"].to_pylist()) test_doubles_are_big() ``` ## Gotchas * **Windows need a key for facts about order.** `LAG`, `LEAD` and `ROW_NUMBER` depend on how ties break unless `requires` has a key on the partition and order columns (`sv.unique("device_id", "ts")`). Without it `verify` proves only what holds for every order of the ties (D24), `report.deterministic` is False and `report.tie_places` names the window; a rerun that must give the same rows refuses the plan. * **SQL strings are DuckDB's SQL**, read by sverify's lift, which covers a subset so far (SELECT, WHERE, joins, subqueries and CTEs, GROUP BY with grouping sets, windows, set operations, ORDER BY, LIMIT, CASE, CAST and common functions): write the rest in Ibis. In DuckDB's SQL, `/` divides as doubles; integer division is `//`. Checks use sverify's reference semantics unless you pass `semantics="duckdb"` (to `sv.sql`, `sv.verify` or `@sv.transform`), which models DuckDB's NaN and `±inf`. * **Counts across rows are not tracked**, so `COUNT(*) >= 1` per group and relations beyond "column plus constant" (`a < 2 * b`) aren't proved. * **Incremental maintenance** covers filters, projections, inner and cross joins and `UNION ALL`; aggregates and windows recompute only the groups and partitions a batch touches; other nodes (left joins, `DISTINCT`, `LIMIT`) are recomputed and diffed on each refresh (see the `sverify-pipelines` skill). * **Sharded `SUM`/`AVG` of doubles is refused** (`sverify.engines.run_sharded`), since the rounding would depend on the split; `float_order="any"` splits them anyway, within a proved tolerance that `tolerance=True` reports per row. ## More * [references/contracts.md](references/contracts.md): what Lean tracks, SQL transformations, composing contracts. * The `sverify-pipelines` skill: Delta and Iceberg sources, sinks, refresh. * The `sverify-engines` skill: running a plan on DuckDB, Polars, DataFusion, Snowflake or Spark, and ASOF joins.
references/contracts.md
# Contracts in more depth ## What Lean tracks Per output column: whether it can be NULL, integer, double and timestamp ranges, which booleans it can be, and whether it can be a string. Filters (`amount > 10`, `x IN (1, 2, 3)`, `x IS NOT NULL`, `x IS NULL OR x >= 0`) refine those facts; arithmetic, `CASE`, `COALESCE`, joins and `UNION ALL` carry them through. A non-nullable schema column is never NULL. Join and filter conditions such as `a.ts = b.ts`, `t0 < t1`, `w1.ts = w0.ts + 60` or `value != -9999` become facts about each row, so postconditions like `first_txn < duplicate_txn` follow. Keys survive filters, projections, aggregates (the grouping columns are a key) and joins whose condition pins down the other side's key. Doubles compare as SQL engines compare them: one zero, and one NaN, equal to itself and above every number, so `NaN > 30` is TRUE and a filter `v > 30` keeps NaN rows. Ask for `~_.v.isnan()` if a NaN must not get through. Not tracked: string contents, relations beyond "column plus constant", counts across rows. ## SQL transformations `sv.verify` takes a SQL string (DuckDB's SQL) with the input schemas: ```python import ibis import sverify as sv orders = ibis.table({"id": "int64", "amount": "int64"}, name="orders") report = sv.verify( "select id, amount * 2 as doubled from orders where amount > 10", tables={"orders": orders}, requires={"orders": ["amount >= 0", "amount < 1000000000", sv.unique("id")]}, ensures=["doubled >= 22", sv.unique("id")], ) assert report.proved ``` In a node, write the SQL as a transform's body, `return sv.sql(query, orders=orders)`: sverify proves, composes and refreshes it like an Ibis transform. ## Composing contracts When one contracted step feeds another, the downstream preconditions are proved from the upstream contract instead of being checked on the data. A broken link raises `sverify.errors.CompositionError`, when the node is built (on sources) or the expression is (on Ibis tables): ```python from ibis import _ @sv.transform( requires={"orders": [_.amount >= 0, _.amount < 1_000_000_000, sv.unique("id")]}, ensures=[_.amount >= 0, _.amount < 1_000_000_000, _.amount.notnull(), sv.unique("id")], ) def paid(orders): return orders.filter(orders.amount.notnull()) @sv.transform( requires={"t": [_.amount >= 0, _.amount < 1_000_000_000, _.amount.notnull()]}, ensures=[_.doubled >= 0], ) def doubled(t): return t.select(t.id, doubled=t.amount * 2) expr = doubled(paid(orders)) print(doubled.report) # requires paid: ... (what `paid` guarantees) ``` `sv.contracts.provenance(expr)` shows what a contracted result assumed and guarantees. ASOF joins (`sv.asof_join`) are in the `sverify-engines` skill.
SKILL LAYOUT
sverify-pipelines/
├── SKILL.md
└── references/
├── custom-io.md
└── table-formats.md
SKILL.md
--- name: sverify-pipelines description: > Build incrementally maintained, contract-carrying data jobs over Delta Lake or Iceberg tables with sverify. Use when a user wants to keep a query result up to date from a table's change feed instead of recomputing it, replace a watermark ("rows newer than…") job, store results in a Delta or Iceberg table exactly once, chain jobs through stored tables, resume a restarted job, prove preconditions from table metadata, read from an API or a folder of files, write to a sink of their own, or set up credentials. --- # sverify jobs: Delta and Iceberg, kept up to date A contracted transform (`@sv.transform`) called on a source builds a node; transforms called on nodes compose. `node.write(sink)` stores a node, and `node.refresh()` brings its whole lineage up to date, upstream first. Nodes in the incremental fragment (filters, projections, inner and cross joins, `UNION ALL`) are updated by delta plans proved equal to a full recompute; aggregates and windows recompute only the groups and partitions a batch touches; others (left joins, `DISTINCT`, `LIMIT`) are recomputed and diffed. Needs the `delta` or `iceberg` extra. A node whose contract is not proved refuses the refresh (`UnverifiedRunError`, a `VerificationError`, "not verified: …"); `refresh(unverified=True)`, or `on_unproved="warn"` on its transform, runs it anyway and records the route `"unverified"`. A sink accepts doubles within the plan's proved bound by default (`sv.equivalent(floats="plan")`); `reproducible=sv.equivalent()` asks for the same bits on a rerun. ## Source, contracted transforms, sink ```python import datetime as dt from pathlib import Path from tempfile import mkdtemp import pyarrow as pa from deltalake import DeltaTable, write_deltalake from ibis import _ import sverify as sv from sverify.io import DeltaSink, DeltaSource lake = Path(mkdtemp()) SCHEMA = pa.schema( [ pa.field("id", pa.int64(), nullable=False), pa.field("amount", pa.int64(), nullable=False), pa.field("ts", pa.timestamp("us"), nullable=False), ] ) def rows(*rs): return pa.Table.from_pylist([dict(zip(SCHEMA.names, r)) for r in rs], schema=SCHEMA) write_deltalake( lake / "orders", rows((1, 30, dt.datetime(2026, 1, 1)), (2, 5, dt.datetime(2026, 1, 2))), configuration={"delta.enableChangeDataFeed": "true"}, # refresh reads the CDF ) DeltaTable(lake / "orders").alter.add_constraint({"non_negative": "amount >= 0"}) @sv.transform( requires={"orders": [_.amount >= 0, _.amount < 1_000_000_000, sv.unique("id")]}, ensures=[_.doubled >= 22, sv.unique("id")], ) def big(orders): return orders.filter(orders.amount > 10).select(orders.id, doubled=orders.amount * 2) orders = DeltaSource(str(lake / "orders")) stored = big(orders).write(DeltaSink(str(lake / "big"))) # proved here first = stored.refresh() # checks preconditions, computes, writes the sink for p in first["big"].preconditions: print(p) # orders: unique(id) (checked on data) # orders: (_.amount >= 0) (Delta constraint) ``` The sink's table gets the proved guarantees as NOT NULL columns and CHECK constraints, and the contract in the commit. A transform called on `stored` reads the table as a versioned source and builds on that contract. ## Refresh from the change feed ```python t = DeltaTable(lake / "orders") write_deltalake(t, rows((3, 50, dt.datetime(2026, 1, 3))), mode="append") t.delete("id = 1") print(stored.refresh(audit=True)) # CDF v1→v3, delta plans, one MERGE commit # big: # orders@v1→v3: 1 insert(s), 1 delete(s); Δ trusted as coming from Delta CDF v1→v3 # ... assert stored.to_arrow().to_pylist() == [{"id": 3, "doubled": 100}] ``` * `refresh()` is all or nothing: a batch that breaks a precondition raises `sverify.errors.PreconditionError` and changes nothing (sinks included); the next refresh retries. * `audit=True` also compares the batch with the new snapshot and every node with a recompute before committing (`sverify.errors.AuditError`). * `refresh(engine="polars")` (or `"datafusion"`, `"spark"`, a Spark session) picks the engine at refresh time. * `refresh().summary()` gives one line per node for the log of a job that refreshes often; `str(refresh)` is the full report. * Where the change feed is gone (CDF off, files vacuumed), the changed files are diffed, and the report says so. ## Restart a job, and refresh several outputs together A restarted job builds the same nodes again; the first refresh resumes each stored node from its sink's table instead of recomputing it, if the plan and source versions match what the sink recorded: ```python restarted = big(DeltaSource(str(lake / "orders"))).write(DeltaSink(str(lake / "big"))) print(restarted.refresh()) # big: # resumed from .../big@v3 # ... assert restarted.to_arrow().to_pylist() == [{"id": 3, "doubled": 100}] ``` `sv.group(a, b).refresh()` refreshes several outputs against the same source versions, sharing the stored nodes they read; `.status()` reports on each. A sink that writes every minute gathers commits and small files. Give the `DeltaSink` its maintenance: ```python minutely = DeltaSink( str(lake / "big_minutely"), checkpoint_interval=10, log_retention=dt.timedelta(days=7), compact_every=60, vacuum_retention=dt.timedelta(days=1), ) ``` `compact_every=60` compacts after every 60 writes, and `vacuum_retention` then removes the files no version from the last day needs; `sink.maintain()` does both by hand. The maintenance commits carry the contract. A restarted job resumes past them, and past another writer's `OPTIMIZE`, `VACUUM` or property change (an Iceberg sink past `replace` snapshots). ## Gotchas * Turn on Delta's Change Data Feed (`delta.enableChangeDataFeed`) or every refresh falls back to diffing. * Keys are always checked on the data (neither format enforces them); other preconditions come from schema nullability, CHECK constraints and file statistics where possible. * A sink needs a version for every input: wrap an in-memory lookup table in `sverify.io.StaticSource(table, version=1, name="regions")`. * Without a `.write` between transforms, they fuse into one plan and the intermediate rows are never stored. Write a node when another job reads it, or when recomputing it would be expensive. * A node with a contract isn't re-planned when a source's schema changes incompatibly: `refresh` raises `sverify.errors.SchemaError` and changes nothing. ## More [references/table-formats.md](references/table-formats.md): Iceberg, file pruning, static sources, chaining jobs through stored tables. [references/custom-io.md](references/custom-io.md): your own sources (an API, files, `from_query`, `from_frame`) and sinks (`Sink`, `sink_from`), and credentials for S3, GCS, Azure and Iceberg catalogs (`Storage`).
references/custom-io.md
# Your own sources and sinks, and credentials Read from an API, a database or a folder of files, and write to anything, with the same exactly-once writes and contracts as Delta and Iceberg. ## A source: `read(since)` returns new rows and a cursor `from_query(fn, cursor_on=...)` covers the high-water-mark case: `fn(since)` returns the rows whose `cursor_on` is above `since` (all of them when `since` is None), and the next read starts from the greatest value read. ```python import pyarrow as pa import pyarrow.compute as pc from ibis import _ import sverify as sv from sverify.io import from_query events = pa.table({"id": [1, 2], "ts": [10, 20], "amount": [30, 5]}) def events_after(since): return ( events if since is None else events.filter(pc.call_function("greater", [events["ts"], since])) ) @sv.transform(requires={"events": [_.amount >= 0]}, ensures=[_.amount > 10]) def big(events): return events.filter(events.amount > 10) node = big(from_query(events_after, cursor_on="ts", name="events")) node.refresh() events = pa.concat_tables([events, pa.table({"id": [3], "ts": [30], "amount": [50]})]) print(node.refresh()) # big: # events@v1→v2: 1 insert(s), 0 delete(s); Δ trusted as rows appended to events v1→v2, assuming rows only arrive with a greater ts # ... ``` Each read that changes something is a new version. The report repeats what the batches assume; `refresh(audit=True)` reads `fn(None)` up to the cursor and raises `sverify.errors.AuditError` where a row arrived late, below the cursor. Other helpers: * `from_frame(fn, name=...)`: `fn()` returns the whole table on every refresh; each read is diffed against the last, so the change is exact. * `from_files(path, format="parquet", name=...)`: the files new in a directory (local, `s3://`, ...) since the last refresh, appended. For anything else, subclass `Source`. `read` returns a `Batch` (an Arrow table, a Polars or pandas DataFrame) with a cursor and a kind: `"append"` (new rows), `"snapshot"` (the whole table, diffed) or `"changes"` (rows added, and `deletes=` rows removed): ```python from sverify.io import Batch, Source class Events(Source): """An append-only log read by offset: the offset lasts across restarts.""" def __init__(self, log): super().__init__("events", assumes="events are only appended") self.log = log def read(self, since): new = self.log[since or 0 :] return Batch(pa.Table.from_pylist(new), cursor=len(self.log)) if new else None def restore(self, cursor): if cursor > len(self.log): return None # truncated: read everything again return pa.Table.from_pylist(self.log[:cursor]) class Prices(Source): """A price list fetched whole: the source diffs each fetch.""" def __init__(self, fetch): super().__init__("prices") self.fetch = fetch def read(self, since): return Batch(self.fetch(), kind="snapshot") prices = Prices(lambda: pa.table({"sku": ["a", "b"], "price": [3, 4]})) print(prices.latest(), prices.snapshot().table.num_rows) # 1 2 ``` What such sources can and cannot promise: * They have no metadata, so preconditions on them are checked on the rows. * They keep the table as of the latest read in memory (joins and keys need it). * A sink records each source's version and cursor. A source that implements `restore(cursor)`, the whole table as of a recorded cursor, resumes there after a restart: the stored nodes resume from their sinks and the first refresh reads `read(cursor)`. `from_query` and `from_files` implement it. The cursor is recorded if it has a JSON form (datetimes, dates, decimals, bytes and tuples are tagged). * A source without `restore`, or whose `restore` returns None (the cursor's rows are gone), is read in full again. Its versions continue above the ones the job's sinks recorded, so the stored nodes reading it are recomputed and written once, never skipped by mistake. * Give `schema=` too: building a node needs the schema, and without it a restarted source reads the whole table once before it restores. * A table format with statistics and a change feed implements the full `sverify.io.VersionedSource` protocol instead (`schema`, `latest`, `snapshot`, `changes`), building its results with `Snapshot` and `SourceBatch.unchanged`, `.from_changes` and `.from_snapshot`. * An Arrow Flight server is a source already: `sverify.io.flight.FlightSource(location, "samples")` (the `flight` extra) reads the batches appended since the last refresh, assuming the server numbers them in order of arrival. `schema=` proves contracts without asking the server, and `token=` sends a bearer token. ## A sink: `write(change)`, and `contract()` for exactly once `Sink.write` gets a `SinkChange`: `inserts` and `deletes` since the last write (Arrow tables), or with `replace` the whole `table`, and the `contract` to record. sverify skips a state whose `contract.epoch` the sink has already committed, unless another plan committed it (`contract.plan` differs) or it was computed from other tables (a source dropped and created again: a lower version, or another id in `contract.table_ids`): then it hands `write` the whole table with `replace`. `contract()` reads the last contract back from the sink, so that holds across restarts too; without it, sverify remembers for the life of the process. If two processes may refresh the same sink, make `write` fail for the second one to commit: raise `sverify.errors.ConcurrentWriteError` when the sink no longer holds what `contract()` returned. Its next refresh writes the whole table, or skips the state the other process committed. ```python import json from pathlib import Path from sverify.io import Contract, Sink, StaticSource class JsonLines(Sink): """Each state as a JSON lines file, with the contract beside it.""" 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): rows = change.table.to_pylist() self.path.write_text("".join(json.dumps(r) + "\n" for r in rows)) self.record.write_text(change.contract.to_json()) orders = StaticSource(pa.table({"id": [1, 2], "ts": [1, 2], "amount": [30, 5]}), name="events") stored = big(orders).write(JsonLines("big.jsonl")) print(stored.refresh().writes[0]) # big from events@v0: overwrite, 1 row(s) written, 0 deleted print(big(orders).write(JsonLines("big.jsonl")).refresh().writes[0]) # a restarted job # big from events@v0: already committed ``` * `sink_from(fn, name=...)` hands the whole table to `fn` whenever it changes: `node.write(sink_from(lambda t: t.to_pandas().to_csv("out.csv"), name="out"))`. * `change.schema` has the columns the node proves NOT NULL as not nullable. Store `change.contract.checks` as constraints where the target has them. * Nodes reading a stored node read what this process wrote, with its exact changes; override `source()` to have them read the sink instead. * `write` may return a `sverify.io.SinkWrite` to report its own mode and version; None reports `"overwrite"` or `"update"`. ## Credentials and storage sverify stores no credentials: it passes options to the library doing the I/O, and the contract a sink records holds none. * **Delta** (S3, GCS, Azure): delta-rs's default chain applies when no options are given: environment variables (`AWS_REGION`, `AWS_ACCESS_KEY_ID`, ...), `~/.aws` profiles, instance or pod roles. Prefer it. Otherwise give `storage_options`: a `sverify.io.Storage` shared by every table, holding a mapping or a function that returns fresh options, called whenever a table is opened (so expiring tokens renew). `repr` shows option names only. * **Iceberg** (REST, Glue, ...): the pyiceberg catalog holds them, from `load_catalog` properties, `~/.pyiceberg.yaml` or `PYICEBERG_*` variables. * **Files** (`from_files`): the filesystem `path`'s URI names, with its default chain, or `filesystem=` a `pyarrow.fs.FileSystem`. * **Your own source or sink**: take a `Storage` and call `options()` when you connect, rather than literal keys, which travel with any pickled object. ```python from sverify.io import DeltaSink, DeltaSource, Storage def s3_options(): # Fetch short-lived credentials here, e.g. from boto3 or a vault. return {"AWS_REGION": "eu-central-1"} s3 = Storage(s3_options) orders_s3 = DeltaSource("s3://lake/orders", storage_options=s3) # nothing opened yet paid_s3 = DeltaSink("s3://lake/paid", storage_options=s3) print(s3, Storage({"AWS_SECRET_ACCESS_KEY": "..."})) # Storage(s3_options) Storage(AWS_SECRET_ACCESS_KEY) ``` ```yaml # ~/.pyiceberg.yaml: an Iceberg REST catalog with a token catalog: lake: uri: https://catalog.example.com token: <from your secret store> ```
references/table-formats.md
# Table formats in more depth ## Iceberg `IcebergSource(catalog, "db.table")` and `IcebergSink` work like their Delta counterparts. Versions are snapshot sequence numbers; a run of appends is read from the files those snapshots added, and anything else (row-level deletes, overwrites) is a diff of the data files that changed. Iceberg has no CHECK constraints, so the sink keeps the contract in the table properties and snapshot summaries. Tables with equality deletes can't be read. ```python import datetime as dt import tempfile import warnings import pyarrow as pa from ibis import _ from pyiceberg.catalog.sql import SqlCatalog import sverify as sv from sverify.io import IcebergSink, IcebergSource warnings.filterwarnings("ignore", module="pyiceberg") d = tempfile.mkdtemp() catalog = SqlCatalog("lake", uri=f"sqlite:///{d}/catalog.db", warehouse=f"file://{d}") catalog.create_namespace("plant") READING = pa.schema( [ pa.field("device_id", pa.int64(), nullable=False), pa.field("ts", pa.timestamp("us"), nullable=False), pa.field("temperature", pa.float64()), ] ) def readings(*rs): return pa.Table.from_pylist([dict(zip(READING.names, r)) for r in rs], schema=READING) table = catalog.create_table("plant.readings", schema=READING) table.append( readings( (1, dt.datetime(2026, 7, 1, 12, 0), 31.0), (2, dt.datetime(2026, 7, 1, 12, 0), float("nan")) ) ) @sv.transform( requires={"readings": [sv.unique("device_id", "ts")]}, ensures=[_.temperature > 30, ~_.temperature.isnan()], ) def hot(readings): t = readings.temperature return readings.filter((t > 30) & ~t.isnan()) hot_readings = hot(IcebergSource(catalog, "plant.readings", name="readings")) stored = hot_readings.write(IcebergSink(catalog, "plant.hot")) stored.refresh() table.append(readings((1, dt.datetime(2026, 7, 1, 12, 1), 33.0))) print(stored.refresh()) # hot: # readings@v1→v2: 1 insert(s), 0 delete(s); Δ trusted as the files Iceberg appends v1→v2 added # ... assert len(stored.to_arrow()) == 2 ``` `NaN > 30` is TRUE in SQL (NaN sorts above every number), which is why the filter and the postcondition rule NaN out explicitly. ## Proved file pruning (Delta) `DeltaSource(uri, where="amount * 2 > 100")` reads only the rows on which the condition is TRUE, and skips a file only if Lean proves from its statistics that no row of it can match. Changes are filtered the same way, and the condition becomes a fact the contracts can use. ## Tables outside a table format A sink writes exactly once per state, keyed by the versions of all its inputs. A small lookup table that lives in memory becomes a source with a version you choose; `apply` changes some of its rows, and `replace` swaps the whole table (the next refresh diffs it): ```python from sverify.io import StaticSource regions = StaticSource( pa.table({"device_id": [1, 2], "region": ["north", "south"]}), version=1, name="regions" ) @sv.transform() def located(readings, regions): return readings.join(regions, readings.device_id == regions.device_id).select( readings.ts, regions.region ) where = located(IcebergSource(catalog, "plant.readings", name="readings"), regions) where.refresh() regions.replace(pa.table({"device_id": [1, 2], "region": ["north", "east"]}), version=2) where.refresh() # recomputes from the new regions assert sorted(set(where.to_arrow().column("region").to_pylist())) == ["east", "north"] ``` ## Chaining jobs through stored tables A transform called on a written node reads its table as a versioned source and builds on the contract the sink stored (Delta: NOT NULL columns and CHECK constraints), instead of checking the data again: `revenue(paid(orders).write(DeltaSink(...))).write(DeltaSink(...))`. A separate job that opens the table with `DeltaSource(uri)` gets the same: `DeltaSource(uri).contract()` returns the contract, and each refresh report names where every precondition came from.
SKILL LAYOUT
sverify-engines/
├── SKILL.md
└── references/
└── semantics.md
SKILL.md
--- name: sverify-engines description: > Run sverify-checked transforms on DuckDB, Polars, DataFusion or Spark, or write their plans as DuckDB, Snowflake or Spark SQL with the same semantics. Use when a user wants the same result on several engines, moves a job between engines or to a warehouse, gets `sverify.engines.Unsupported` and wants to know why, needs `deterministic=True`, or aligns two time series with `asof_join`. --- # sverify engines: one plan, the same answer everywhere The same SQL text means different things on different engines: integer `/` truncates in DataFusion and PostgreSQL but gives a double in DuckDB and Spark; DataFusion wraps on integer overflow where DuckDB raises; NULLs sort first or last depending on the engine; ties in a sort or window break differently. sverify translates the plan Lean checked for each engine with every such choice spelled out, or refuses it where an engine can't match. ## Pick the engine when you refresh A node runs the plan Lean checked; the engine is chosen at refresh time, and the rows are the same on each: ```python import json import ibis import pyarrow as pa import sverify as sv from sverify.engines import ENGINES from sverify.io import StaticSource orders = StaticSource(pa.table({"id": [1, 2, 3], "amount": [7, -7, None]}), name="orders") @sv.transform() def halves(orders): return orders.select(orders.id, half=orders.amount // 2) results = {} for engine in ENGINES: # "duckdb", and with their extras "polars" and "datafusion" node = halves(orders) node.refresh(engine=engine) # A node's rows come in no particular order: compare them as bags. results[engine] = sorted(map(repr, node.to_arrow().to_pylist())) print(sorted(results)) # ['datafusion', 'duckdb', 'polars'] assert len({repr(r) for r in results.values()}) == 1 # the same rows everywhere ``` `refresh(engine="spark")` or a Spark session runs on Spark (the `spark` extra, Java 17 or 21). `node.to_polars()` and `node.to_pandas()` return the rows in those libraries. ## Write the plan as warehouse SQL `node.to_sql(dialect)` writes the node's plan as SQL whose meaning is the model's, and refuses a plan whose contract is not proved; `report.to_sql` does the same for a report (`report.to_substrait()` is the portable plan): ```python for dialect in ("duckdb", "snowflake", "spark"): assert node.to_sql(dialect).startswith("SELECT") ``` `sverify.engines.run(con, node.internal_plan, engine)` runs the plan over a connection's tables; `engine="spark"` or a Spark session configured with `SPARK_CONF` runs it on Spark. `sverify.engines.to_sql(plan, dialect)` writes any plan in sverify's internal form, and is outside the 1.x promise. ## Why a plan is refused `to_sql` and the engines raise `Unsupported` (with the reason) instead of answering differently from the model: ```python from sverify.engines import Unsupported @sv.transform() def numbered(orders): w = ibis.window(group_by="id", order_by="amount") return orders.mutate(rn=ibis.row_number().over(w)) try: numbered(orders).to_sql("snowflake") except Unsupported as e: print(str(e).split(",")[0]) # Snowflake tables have no row order @sv.transform(requires={"orders": [sv.unique("id")]}) def numbered_by_id(orders): w = ibis.window(group_by="id", order_by="amount") return orders.mutate(rn=ibis.row_number().over(w)) print(numbered_by_id(orders).to_sql("snowflake")[:6]) # no ties: the report is deterministic # SELECT ``` * **Snowflake and Spark tables have no row order**, so sorts, `LIMIT` and windows are refused unless the report is `deterministic` (a key on the partition and order columns rules out ties), and so is `SUM` or `AVG` of doubles (the rounding depends on the order of additions). `node.to_sql` and `report.to_sql` read it from the report; `sverify.engines.to_sql` takes `deterministic=True`, which you pass only when a report says so. * **Snowflake has no interval type**: interval-valued expressions are refused. * **Warehouse setup the SQL assumes:** Snowflake time columns are `TIMESTAMP_NTZ`, strings have no collation, names are quoted lower case; Spark sessions use `sverify.engines.SPARK_CONF` (ANSI mode, a pinned time zone, NTZ timestamps) and the `UTF8_BINARY` collation. ## Align two time series: `asof_join` `sv.asof_join(left, right, on="ts", by="device", tolerance=...)` matches each left row with the latest right row at or before it, at most `tolerance` earlier, built from a range join and `ROW_NUMBER` that Lean checks like any plan. Without a key on `(by, on)` on the right side, two right rows may tie for "latest": the guarantees hold for either pick, and the result is not deterministic. With keys on both sides it is: ```python from datetime import datetime from ibis import _ acc = ibis.table({"device": "!int64", "ts": "!timestamp", "ax": "float64"}, name="acc") gyr = ibis.table({"device": "!int64", "ts": "!timestamp", "gx": "float64"}, name="gyr") @sv.transform( requires={ "acc": [sv.unique("device", "ts"), _.ts >= datetime(2000, 1, 1)], "gyr": [sv.unique("device", "ts")], }, ensures=[sv.unique("device", "ts"), _.ts_right.isnull() | (_.ts_right <= _.ts)], ) def aligned(acc, gyr): return sv.asof_join(acc, gyr, on="ts", by="device", tolerance=ibis.interval(milliseconds=10)) aligned(acc, gyr) # on Ibis tables: checks the contract, returns an expression assert aligned.report.proved ``` ## More [references/semantics.md](references/semantics.md): how each engine is made to agree with the model (NULLs, integers, doubles, strings, time).
references/semantics.md
# How the engines are made to agree The checked plan has one meaning, `Rel.eval` in the Lean model. Each translation writes that meaning out explicitly, and raises or refuses where an engine can't follow it. The differential tests compare every engine with the model on random tables. * **NULL logic.** Filters and join conditions keep a row only when the condition is TRUE (`IS TRUE`); `AND`/`OR` are three-valued; equi-join keys never match on NULL. Every `ORDER BY` says where NULLs go. * **Integers** have their declared widths. Overflow raises on DuckDB, Polars and Spark (ANSI mode), and Snowflake's NUMBER(38) is range-checked, where DataFusion wraps silently. Integer division and remainder truncate toward zero everywhere. Division by zero, `MIN / -1`, bad `POWER`s and out-of-range casts raise. * **Doubles** have one zero (−0.0 is 0.0) and one NaN, equal to itself and above every number, in comparisons, joins, grouping and sorts. `SUM` and `AVG` of doubles add in the model's order on DuckDB and Polars; on Snowflake and Spark, whose tables have no order, they are refused. * **Order.** Below a `LIMIT`, a window or a sort, an ordinal column carries the model's row order, so ties break the same way on DuckDB and Polars. Snowflake and Spark need `deterministic=True`. * **Strings** count characters, not bytes, in lengths, positions and `substring`; `LIKE` names its escape character. * **Time.** Timestamps are microseconds without a time zone (`TIMESTAMP_NTZ` on Snowflake and Spark); `date_part` and `date_trunc` follow the model's calendar (ISO weeks), not the session's week start. DataFusion runs the plan through its own Substrait consumer, so its own semantics apply there; section 8 of `catalogue/backend_semantics.py` (one expression, many engines) lists where they differ from the model.