Delta Lake¶

The DeltaIOHandler persists asset outputs as Delta Lake tables. It supports PyArrow, Polars, pandas, and DataFusion data types, partitioned writes, and full MERGE INTO operations.

Setup¶

pip install rivers[delta]

# Plus at least one of:
pip install rivers[pyarrow]
pip install rivers[polars]
pip install rivers[pandas]
pip install rivers[datafusion]
pip install rivers[duckdb]
pip install rivers[pyspark]

Basic usage¶

import polars as pl
import rivers as rs
from rivers.io_handlers.delta import DeltaIOHandler

io = DeltaIOHandler(table_uri="/data/delta")

@rs.Asset(io_handler=io)
def users() -> pl.DataFrame:
    return pl.DataFrame({
        "id": [1, 2, 3],
        "name": ["Alice", "Bob", "Carol"],
    })

Usage with Spark¶

import pandas as pd
from pyspark.sql.dataframe import DataFrame as SparkDataFrame
import rivers as rs
from rivers.io_handlers.delta import DeltaIOHandler

spark_session = ... # your spark session initialization
io = DeltaIOHandler(
    table_uri="/data/delta",
    handler_config={"spark_session": spark_session}
)

@rs.Asset(io_handler=io)
def users() -> SparkDataFrame:
    return spark.createDataFrame(
        pd.DataFrame({
            "id": [1, 2, 3],
            "name": ["Alice", "Bob", "Carol"],
        })
    )

In above cases, the handler creates a Delta table at /data/delta/users/.

Supported types¶

Type	Extra	IO Type
`pyarrow.Table`	`rivers[pyarrow]`	Arrow
`pyarrow.RecordBatchReader`	`rivers[pyarrow]`	Arrow
`polars.DataFrame`	`rivers[polars]`	Arrow
`polars.LazyFrame`	`rivers[polars]`	Arrow
`pandas.DataFrame`	`rivers[pandas]`	Arrow
`datafusion.DataFrame`	`rivers[datafusion]`	Arrow
`duckdb.DuckDBPyRelation`	`rivers[duckdb]`	Arrow
`pyspark.sql.DataFrame`	`rivers[pyspark]`	Spark

The type is detected automatically from the object passed to handle_output, and load_input uses the type_hint from the downstream parameter annotation.

When read back as a datafusion.DataFrame, the Delta table is registered with a DataFusion SessionContext and returned as a lazy query — column projection and the partition predicate are pushed into the scan, and execution happens when the consumer collects or streams it (like a polars.LazyFrame).

import datafusion
import rivers as rs

@rs.Asset
def enriched(users: datafusion.DataFrame, orders: datafusion.DataFrame) -> datafusion.DataFrame:
    return users.join(orders, on="id", how="inner")

The backing SessionContext is attached to the returned frame as rivers_ctx. Reach for it only when you want the session handle itself — for example, to register an extra table and query it by name with ctx.sql(...).

Read with DuckDB¶

A duckdb.DuckDBPyRelation input reads the table with DuckDB's delta_scan. The relation is lazy. delta/columns, the partition predicate, and file statistics prune the files that DuckDB reads, and delta/version reads an older version. A relation output streams into delta-rs, which writes the table.

import duckdb
import rivers as rs

@rs.Asset
def summary(orders: duckdb.DuckDBPyRelation, users: duckdb.DuckDBPyRelation) -> duckdb.DuckDBPyRelation:
    return orders.join(users, "user_id").aggregate("country, sum(amount) AS total")

Inputs that one step reads share one DuckDB connection per thread, so the step can join them. To set extensions, secrets, or settings, pass a DuckDBResource:

from rivers.integrations.duckdb import DuckDBResource

io = DeltaIOHandler(
    table_uri="s3://bucket/delta",
    storage_options={"AWS_REGION": "eu-west-1"},
    handler_config={"duckdb": DuckDBResource(connection_config={"threads": 8})},
)

The handler turns storage_options into a DuckDB secret for the table's bucket or container:

S3: the access key, secret, session token, region, and endpoint. Without a key, DuckDB uses the AWS credential chain.
Azure: an account key, a SAS token, or a service principal. Without one, DuckDB uses the Azure credential chain.
GCS: not supported. DuckDB reads GCS only with HMAC keys; set them with DuckDBResource(secrets=...).

Write modes¶

Set the default mode on the handler, or override per-asset via metadata:

# Handler-level default
io = DeltaIOHandler(table_uri="/data/delta", mode="append")

# Per-asset override
@rs.Asset(io_handler=io, metadata={"delta/mode": "overwrite"})
def events() -> pl.DataFrame:
    ...

Arrow type handlers¶

Mode	Behavior
`overwrite`	Replace the table (or partition)
`append`	Add rows to existing table
`error`	Fail if table exists
`ignore`	Skip write if table exists
`merge`	MERGE INTO (see below)
`create_or_replace`	Drop and recreate schema, then append

Spark type handlers¶

Supported table write modes¶

Mode	Behavior
`overwrite`	Replace the table (or partition)
`append`	Add rows to existing table
`error`	Fail if table exists
`ignore`	Skip write if table exists

Support schema modes¶

Mode	Behavior
`overwrite`	Replace the table (or partition)
`merge`	MERGE INTO (see below)

Partitioned writes¶

Combine with PartitionsDefinition and partition_expr:

@rs.Asset(
    io_handler=io,
    partitions_def=rs.PartitionsDefinition.daily(start=datetime(2024, 1, 1)),
    metadata={"delta/partition_expr": "date"},
)
def daily_events() -> pl.DataFrame:
    ...

When writing in overwrite mode with a partition key, the handler generates a predicate to overwrite only the target partition.

Merge operations¶

For upserts, deduplication, and other MERGE INTO patterns:

from rivers.io_handlers.delta import DeltaIOHandler, MergeConfig

io = DeltaIOHandler(
    table_uri="/data/delta",
    mode="merge",
    merge_config=MergeConfig(
        merge_type="upsert",
        predicate="s.id = t.id",
    ),
)

Merge types¶

Type	Behavior
`upsert`	Update matched rows, insert unmatched
`deduplicate_insert`	Insert only unmatched rows
`update_only`	Update matched rows only
`replace_delete_unmatched`	Update matched, delete unmatched in target
`custom`	Full control via `MergeOperationsConfig`

Custom merge operations¶

from rivers.io_handlers.delta import (
    MergeConfig,
    MergeOperationsConfig,
    WhenMatchedUpdateAll,
    WhenNotMatchedInsertAll,
)

io = DeltaIOHandler(
    table_uri="/data/delta",
    mode="merge",
    merge_config=MergeConfig(
        merge_type="custom",
        predicate="s.id = t.id",
        operations=MergeOperationsConfig(
            when_matched_update_all=[
                WhenMatchedUpdateAll(predicate="s.updated_at > t.updated_at"),
            ],
            when_not_matched_insert_all=[
                WhenNotMatchedInsertAll(),
            ],
        ),
    ),
)

Storage options¶

Pass storage credentials for remote backends:

io = DeltaIOHandler(
    table_uri="s3://my-bucket/delta",
    storage_options={
        "aws_region": "us-east-1",
        "aws_access_key_id": "...",
        "aws_secret_access_key": "...",
    },
)

Handler configuration¶

Handler configuration allows you to pass custom handler properties like a SparkSession object when using a Spark type handler.

Example reading `my_table` in a cloud blob storage with Spark¶

spark_session = ... # your spark session initialized with cloud credentials
io_spark = DeltaIOHandler(
    table_uri="/cloud/path/to/my_table",
    handler_config={"spark_session": spark_session}
)
ctx = rs.InputContext(
    asset_name="tbl",
    downstream_asset="consumer",
    type_hint=pyspark.sql.DataFrame,
)
spark_df = handler.load_input(ctx)

Table configuration¶

Set Delta table properties:

io = DeltaIOHandler(
    table_uri="/data/delta",
    table_config={
        "delta.deletedFileRetentionDuration": "interval 30 days",
    },
)

Override per-asset via metadata:

@rs.Asset(
    io_handler=io,
    metadata={
        "delta/table_configuration": '{"delta.enableChangeDataFeed": "true"}',
    },
)
def events() -> pl.DataFrame:
    ...

Reading data¶

Downstream assets read data by type annotation:

@rs.Asset(io_handler=io)
def summary(users: pl.DataFrame) -> pl.DataFrame:
    return users.group_by("region").len()

Column selection¶

Load only specific columns:

@rs.Asset(
    io_handler=io,
    metadata={"delta/columns": '["id", "name"]'},
)
def user_names(users: pl.DataFrame) -> pl.DataFrame:
    ...

Time travel¶

Read a specific table version:

@rs.Asset(io_handler=io, metadata={"delta/version": "3"})
def historical(users: pl.DataFrame) -> pl.DataFrame:
    ...

Maintenance actions¶

Subclass DeltaAsset and the standard maintenance verbs come built in — optimize, vacuum, and delete, resolved against the asset's DeltaIOHandler (same URI, same credentials, same commit configuration the write path uses):

from rivers.io_handlers.delta import DeltaAsset  # also exported as rs.DeltaAsset

class Orders(DeltaAsset):
    io_handler = WAREHOUSE

    @classmethod
    def materialize(cls) -> pl.DataFrame:
        return pl.DataFrame({"id": [1, 2, 3]})


repo = rs.CodeRepository(assets=[Orders])
repo.run_action("optimize")
repo.run_action("optimize", partition_key=pk)  # only that partition's files
repo.run_action("vacuum", config={"orders": {"retention_hours": 24}})
repo.run_action("delete", partition_key=pk)

Verb	Declaration	Behavior
`optimize`	`Unchanged` + `Exclusive` + `Optional` key	Compacts small files; z-orders instead when `z_order_by` is configured. With a key, only that partition's files; without one, the whole table. From the UI or gRPC, the keyless form needs an explicit whole-asset choice.
`vacuum`	`Unchanged` + `Exclusive` + `Keyless`	Removes unreferenced files; `retention_hours` and `enforce_retention_duration` via config. Table-wide — runs without a partition key, on partitioned assets too.
`delete`	`Unmaterialize` + `Exclusive` + `DownstreamFirst` + `Optional` key	Deletes the keyed partition's rows via `partition_predicate`, or every row without a key — both forms valid on partitioned assets. From the UI or gRPC, the keyless form needs an explicit whole-asset choice.

On a table that does not exist yet, optimize and vacuum report unchanged instead of failing, so a fleet-wide repo.run_action("optimize") skips never-materialized assets; delete still applies Unmaterialize, so dangling state clears even when the physical table is already gone. DeltaAsset sets kinds = "delta" and defines no materialize(), so it is a mixin — subclasses supply it, and a subclass redefining a verb replaces it.

Custom verbs¶

For verbs beyond the built-ins, DeltaIOHandler exposes the same per-asset resolution the write path uses, so custom asset actions stay symmetric with writes:

class Events(DeltaAsset):
    io_handler = WAREHOUSE

    @classmethod
    def materialize(cls) -> pl.DataFrame: ...

    @rs.action(outcome=rs.Outcome.MayMaterialize)
    @classmethod
    def merge_late(cls, ctx: rs.ActionContext) -> rs.ActionResult:
        h = ctx.io_handler
        late = fetch_late_rows(ctx.partition_key)
        if late.is_empty():
            return rs.ActionResult.unchanged()
        merge_into(
            h.asset_table_uri(ctx.asset_name, ctx.asset_metadata),
            late,
            storage_options=h.storage_options,
        )
        return rs.ActionResult.materialized(metadata={"rows_merged": len(late)})

asset_table_uri(asset_name, asset_metadata) — {table_uri}/{leaf}, honoring the delta/root_name override.
partition_predicate(asset_metadata, ctx.partition) — the SQL predicate for the partition(s) being acted on, honoring delta/partition_expr.
partition_filters(asset_metadata, ctx.partition) — the same partitions as delta-rs partition_filters, one conjunction per key, for optimize.

Because optimize declares Exclusive, a keyless run claims the asset's implicit pool for the whole asset: it never overlaps a materialize of the same asset, and contention shows up in the UI as StepSlotWaiting. A keyed optimize or delete claims only its partition. Materialize steps never exclude each other, so declaring an exclusive action does not serialize ordinary runs.