Sources

A source produces Arrow batches and a checkpoint per batch. The checkpoint records how far it has read, in whatever shape the source defines.

Which stores read, which only write

Kafka, Delta, Postgres, Databricks, Snowflake, and BigQuery read and write; S3 is sink-only (see Sinks). Each source reads the way that suits the store:

Source	Reads by	Resume checkpoint
Kafka	streaming poll (Rust decode)	per-partition offsets
Delta	table version (append-only or change feed)	{"delta": {"version": N}}
Postgres	incremental query, paginated by a watermark column	{"postgres": {"watermark": v}}
Databricks	a Unity Catalog Delta table, same as Delta	{"delta": {"version": N}}
Snowflake	incremental query, paginated by a watermark column	{"snowflake": {"watermark": v}}
BigQuery	incremental query, paginated by a watermark column	{"bigquery": {"watermark": v}}

Kafka

from pydantic import BaseModel
from pyrula.pipelines import KafkaSource

class Order(BaseModel):
    id: int
    total: float
    note: str | None = None

KafkaSource(consumer_config, value_format="avro", schema=Order,
            max_records=10_000, on_decode_error="halt")

Wraps the consumer’s Rust poll_arrow. value_format is raw, avro, or json, checked at construction. With avro or json you get one Arrow column per field, plus _topic, _partition, _offset, _timestamp, and _key metadata columns.

The schema is an Avro schema for both avro and json (it defines the Arrow column layout). Pass a pydantic model or dataclass and the Avro schema is derived from it (the same model you would hand @kafka_agent), or pass an Avro schema JSON string directly. raw skips decode and gives a single binary value column; it takes no schema.

Checkpoint: {"kafka": {topic: {partition: next_offset}}}.

A record that fails to decode stops the pipeline by default. Set on_decode_error="skip" to seek past the bad offset and keep going. The count is on the source’s skipped.

Delta

from pyrula.pipelines import DeltaSource

DeltaSource("/data/orders", mode="append_only",
            filters=[("region", "=", "us")], columns=["id", "amount"])   # or mode="cdf"

Reads a Delta table incrementally by version. Each poll picks up commits newer than the last one it emitted. append_only reads the files a commit added (inserts); in v1 the table must be on local disk (object-store URIs are not yet supported for this path). cdf uses the Change Data Feed and adds a _change_type column, which needs delta.enableChangeDataFeed=true on the table.

filters= and columns= push the predicate and projection down to the scan, so delta-rs reads fewer files and columns. The filter is a DNF tuple list ([(col, op, val)]), the format delta-rs and pyarrow already speak. Pushdown covers both the snapshot and the incremental added-file reads. CDF mode does not filter the change feed in v1. See Transforms and pushdown.

Checkpoint: {"delta": {"version": N}}.

Postgres

from pyrula.pipelines import PostgresSource

PostgresSource(dsn, table="orders", watermark_column="id", batch_size=10_000,
               filters="region = 'us'", columns=["id", "amount"])

Polls a table on a strictly-increasing column: a serial key or a monotonic timestamp. Each poll runs WHERE wm > $last ORDER BY wm LIMIT batch and fetches the result as Arrow over ADBC, so rows never pass through Python one at a time.

filters= AND-s a SQL fragment into that WHERE clause and columns= projects, so the scan reads less. The fragment is interpolated as operator config, not user data, so never build it from an untrusted caller. The watermark contract is unchanged, so resume still holds. The fragment is raw SQL here; the Delta source takes a DNF tuple list instead. See Transforms and pushdown.

Checkpoint: {"postgres": {"watermark": value}}. Timestamps round-trip as ISO strings.

Databricks (Unity Catalog)

from pyrula.pipelines import DatabricksSource

DatabricksSource("main.sales.events", mode="append_only",
                 filters=[("region", "=", "us")], columns=["id", "amount"])

A Unity Catalog table is Delta underneath, so this is a DeltaSource that first resolves the UC table to its storage path and vends temporary scoped credentials, then reads the object store directly at delta-rs speed (no SQL warehouse in the path). Credentials are re-vended before they expire, so a long run does not die mid-stream. All the Delta knobs apply: mode, filters (DNF tuples), columns, initial_version.

Auth defaults to ambient unified auth, which the SDK resolves to OAuth M2M (service principal), Azure MSI, a profile, or env vars: this is the production path, set DATABRICKS_CLIENT_ID/DATABRICKS_CLIENT_SECRET or a profile in the environment. For other auth, build your own client and pass client=Databricks.from_client(WorkspaceClient(...)). A host=/token= PAT works for local dev but Databricks discourages it for automation. AWS only in v1, and the UC table must be on customer-managed external storage (managed-storage tables cannot vend external credentials). Needs pyrula-pipelines[databricks].

Checkpoint: {"delta": {"version": N}}, same as the Delta source.

Snowflake and BigQuery

from pyrula.pipelines import SnowflakeSource, BigQuerySource

SnowflakeSource(dsn, table="ORDERS", watermark_column="UPDATED_AT", batch_size=10_000)
BigQuerySource(dsn, table="dataset.orders", watermark_column="updated_at")

Both read like the Postgres source: a strictly-increasing watermark column drives WHERE wm > ? ORDER BY wm LIMIT batch, fetched as Arrow over ADBC. They take the same batch_size, columns, and raw-SQL filters= knobs. The checkpoint namespace is the store name ({"snowflake": {...}}, {"bigquery": {...}}); timestamps round-trip as ISO strings. Need pyrula-pipelines[snowflake] or pyrula-pipelines[bigquery].

The Delta, Postgres, Databricks, Snowflake, and BigQuery sources have no source-side commit. They resume from the sink’s checkpoint, so pair them with an exactly-once sink.