io.Source

A table you read yourself: subclass it and implement read.

Usage

io.Source(
    name,
    *,
    schema=None,
    assumes=None,
    on_violation=None,
)

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.

Parameters

name: str: The table’s name in the plans that read it.
schema: pa.Schema | None = None: 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: str | None = None: What the batches rest on, such as "rows are only appended, in order of ts". Default: None.
on_violation: Literal["error", "quarantine", "hold"] | None = None: What a refresh does with a batch that breaks a precondition on this source, for transforms that set no on_violation of their own: "error", "quarantine" or "hold" (see sverify.transform). Default: None, "error".

Attributes

name: str: The table’s name in the plans that read it.
assumes: str | None: What the batches rest on, or None.

Examples

A source reading new rows from a list, by an increasing id:

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

Methods

Name	Description
changes()	The change from version `since` to the latest, as a refresh reads
latest()	Read what is new, and return the latest version.
read()	What is new since the last read.
reread()	The whole table as of `cursor`, read again from scratch, for
restore()	The whole table as of `cursor`, a cursor that a sink recorded in
schema()	The table’s schema: `schema=` without reading, else the first
snapshot()	The table as of a version: the latest read, or an earlier

changes()

The change from version since to the latest, as a refresh reads

Usage

Source

changes(since, until=None)

it: the batches read in between.

Parameters

since: int: The version the node’s state was computed from.
until: int | None = None: The latest version, or None for it.

Returns

SourceBatch: The rows the batches added and removed, or, if they are no
longer kept, the latest table.

Raises

ValueError: until is not the latest version.

latest()

Read what is new, and return the latest version.

Usage

Source

latest()

read()

What is new since the last read.

Usage

Source

read(since)

Parameters

since: Any: The cursor of the last batch read, or None on the first read (and after a batch whose cursor is None).

Returns

Batch | None: The new rows (Batch), or None if nothing is new.

reread()

The whole table as of cursor, read again from scratch, for

Usage

Source

reread(cursor)

refresh(audit=True). Override it to read only up to cursor, so rows that arrive meanwhile are not reported as a mismatch.

Parameters

cursor: Any: The cursor of the latest batch.

Returns

Any: The rows, in any form Batch.table takes. Default: the table of
read(None), or no rows.

restore()

The whole table as of cursor, a cursor that a sink recorded in

Usage

Source

restore(cursor)

an earlier process, so a restarted job resumes from its sinks. Override it for a source whose cursors last: a log’s sequence number, a high-water mark, the files read. Then read(cursor) must return what is new since, and the next refresh is incremental. A sink records the cursor if it has a JSON form (see Batch.cursor).

The restored table is trusted to be the table the stored nodes were computed from; refresh(audit=True) reads the whole table again on later batches.

Parameters

cursor: Any: The cursor the sink recorded, as read returned it (datetimes, dates, decimals, bytes and tuples come back as they were).

Returns

Any

The rows, in any form Batch.table takes, or None if the

cursor is no longer valid (the rows it points to are gone):

then the job reads the whole table, read(None), and recomputes.

Default: Any

None.

schema()

The table’s schema: schema= without reading, else the first

Usage

Source

schema()

batch’s.

snapshot()

The table as of a version: the latest read, or an earlier

Usage

Source

snapshot(version=None)

version whose batches the source still keeps.

Parameters

version: int | None = None: The version, or None for the latest.

Returns

Snapshot: The version and the rows, with no facts.

Raises

ValueError: The source no longer keeps version’s batches, or version is newer than the latest.