io.StaticSource
An in-memory Arrow table as a source, such as a small lookup table or
Usage
io.StaticSource(
table,
*args,
name=None,
version=0,
on_violation=None,
)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.
Parameters
table: pa.Table-
The table’s contents.
name: str | None = None-
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: int = 0-
The version of these contents, by keyword (positional warns for one release). Default: 0.
on_violation: Policy | None = None-
What a refresh does with a batch that breaks a precondition on this source, for transforms that set no
on_violationof 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.
Methods
| Name | Description |
|---|---|
| apply() | Insert and delete rows (an Arrow table, or tuples in schema |
| changes() |
The change from version since to the current one, as a refresh
|
| latest() | The current version. |
| replace() |
Swap in a new table at version, above the current one. The
|
| schema() | The current table’s schema. |
| snapshot() | The current table, as a refresh reads it. |
apply()
Insert and delete rows (an Arrow table, or tuples in schema
Usage
apply(changes, version=None)order), as version (default: one above the current one).
The next refresh of a node reading the source gets exactly this change (several apply calls in between add up), so the node is updated from it rather than recomputed.
Parameters
changes: Changes-
The rows to insert and delete.
version: int | None = None- The new version, above the current one. Default: one above the current one.
Raises
ValueError-
versionis not above the current version, or a deleted row is not in the table (nothing is changed).
Examples
import pyarrow as pa
import sverify as sv
from sverify.io import Changes, 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)
big = big_orders(orders)
big.refresh()
orders.apply(Changes(inserts=[(3, 50)], deletes=[(1, 30)])) # version 1
print(big.refresh())
# big_orders:
# orders@v0→v1: 1 insert(s), 1 delete(s); Δ as applied to orders v0→v1
# view big_orders: orders@v1
assert big.to_arrow().to_pylist() == [{"id": 3, "amount": 50}]changes()
The change from version since to the current one, as a refresh
Usage
changes(since, until=None)reads it.
Parameters
since: int-
The version the node’s state was computed from.
until: int | None = None- The current version, or None for it.
Returns
Raises
ValueError-
untilis not the current version.
latest()
The current version.
Usage
latest()replace()
Swap in a new table at version, above the current one. The
Usage
replace(table, version)next refresh diffs it against the old one, and recomputes the nodes that read the table.
Use apply when you know which rows changed: the refresh then gets exactly that change.
Parameters
table: pa.Table-
The new contents.
version: int- Their version, above the current one.
Raises
ValueError-
versionis not above the current version.
schema()
The current table’s schema.
Usage
schema()snapshot()
The current table, as a refresh reads it.
Usage
snapshot(version=None)Parameters
version: int | None = None- The current version, or None for it.
Returns
Snapshot- The version and the rows, with no facts from metadata.
Raises
ValueError-
versionis not the current one; a static table keeps only its current contents.