Skip to main content

io.from_query()

A source that reads new rows by a high-water mark: fn(since)

Usage

Source

io.from_query(
    fn,
    *,
    cursor_on,
    name,
    schema=None,
)

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.

Parameters

fn: Callable[[Any], Any]

Called with the last cursor, or None; returns the new rows (an Arrow table, a Polars or pandas DataFrame, …) or None.

cursor_on: str

The column whose greatest value is the cursor.

name: str

The table’s name in the plans that read it.

schema: pa.Schema | None = None
The table’s schema, if fn(None) may return nothing. Default: that of the first rows.

Returns

Source
The source.

Examples

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()