DuckDB¶

DuckDBIOHandler¶

Persists asset outputs as tables in a DuckDB database or a DuckLake catalog.

from rivers.integrations.duckdb import DuckDBIOHandler, DuckDBResource

io = DuckDBIOHandler(
    resource=DuckDBResource(database="warehouse.duckdb"),
    mode="overwrite",
)

Constructor:

Supported types:

Lazy inputs read in one thread share DuckDBResource.shared_connection(), so a step can join them.

Asset metadata overrides:

These metadata keys override handler defaults per-asset:

Output metadata:

Methods for action bodies¶

An action receives the asset's resolved handler as ctx.io_handler. These methods expose the same table and partition resolution the write path uses, so a custom action targets exactly the rows the materialize path would have written.

table_name(asset_name, asset_metadata)¶

Returns the (schema, table) pair, honoring duckdb/schema and duckdb/table overrides in asset_metadata. Pass ctx.asset_name and ctx.asset_metadata.

@rs.action(outcome=rs.Outcome.Unchanged)
@classmethod
def analyze(cls, ctx: rs.ActionContext) -> None:
    schema, table = ctx.io_handler.table_name(ctx.asset_name, ctx.asset_metadata)
    with ctx.io_handler.resource.get_connection() as con:
        con.execute(f'ANALYZE "{schema}"."{table}"')

partition_predicate(asset_metadata, partition)¶

Returns a SQL predicate covering the partition(s), honoring duckdb/partition_expr. Pass ctx.asset_metadata and ctx.partition.

@rs.action(outcome=rs.Outcome.Unmaterialize)
@classmethod
def delete(cls, ctx: rs.ActionContext) -> None:
    schema, table = ctx.io_handler.table_name(ctx.asset_name, ctx.asset_metadata)
    predicate = ctx.io_handler.partition_predicate(ctx.asset_metadata, ctx.partition)
    with ctx.io_handler.resource.get_connection() as con:
        con.execute(f'DELETE FROM "{schema}"."{table}" WHERE {predicate}')

partition is required: reach this only from a keyed action run. On a non-partitioned asset, delete the whole table instead of building a predicate.

DuckDBAsset¶

Asset base class with the delete verb built in. Subclass it, define materialize, and delete appears as an action, resolved against the asset's DuckDBIOHandler (its own io_handler or the repository default). Sets kinds = "duckdb".

from rivers.integrations.duckdb import DuckDBAsset

class Orders(DuckDBAsset):
    io_handler = WAREHOUSE

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

A verb requires the asset to resolve a DuckDBIOHandler — anything else fails the step with a TypeError naming the requirement. On a table that does not exist, delete still applies its Unmaterialize outcome, so dangling state clears. A keyed delete on an asset without duckdb/partition_expr fails with a ValueError naming the key. A subclass redefining a verb replaces the built-in.

DuckLakeAsset¶

DuckDBAsset subclass for DuckLake tables, with optimize and vacuum added. Sets kinds = "ducklake".

from rivers.integrations.duckdb import DuckLakeAsset

class Events(DuckLakeAsset):
    io_handler = LAKE
    metadata = {"ducklake/options": '{"rewrite_delete_threshold": 0.5}'}

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

The verbs take no config: set table options with the ducklake/options metadata key, and lake-wide options with DuckLake(options=...). A verb requires the handler's resource to have a DuckLake database — otherwise it fails the step with a TypeError. On a table that does not exist yet, optimize reports ActionResult.unchanged().

DuckDBResource¶

Opens DuckDB connections to a database file, :memory:, MotherDuck, or a DuckLake catalog. Fields are read from DUCKDB_* environment variables when not given.

from rivers.integrations.duckdb import DuckDBResource

warehouse = DuckDBResource(database="warehouse.duckdb", extensions=["spatial"])

Constructor:

Opening a file that another process holds is retried 10 times, with a delay of 0.1 s that doubles each time. After the last try the duckdb.IOException is raised; for a DuckLake catalog in a DuckDB file, the error says that the catalog allows one process.

Methods¶

get_connection()¶

Context manager: a connection that is closed when the block exits. Relations made from it stop working after the block.

with warehouse.get_connection() as con:
    con.execute("CREATE TABLE users AS FROM 'users.parquet'")

connect()¶

Returns a new connection. The caller closes it, or a relation keeps it open. With a DuckLake database, it is a cursor whose default database is the catalog.

shared_connection()¶

Returns the calling thread's connection for relations that must be combined: DuckDB joins relations only when they come from one connection. The connection closes when the last relation made from it is gone.

@rs.Asset(io_handler=io)
def enriched(orders: duckdb.DuckDBPyRelation, warehouse: DuckDBResource) -> duckdb.DuckDBPyRelation:
    regions = warehouse.shared_connection().sql("FROM 'regions.csv'")
    return orders.join(regions, "region_id")

teardown()¶

Releases the DuckLake root connection. The catalog closes when the last cursor or relation made from it is gone.

DuckLake¶

A DuckLake catalog, attached once per process to every connection of a DuckDBResource.

from rivers.integrations.duckdb import DuckDBResource, DuckLake

lake = DuckDBResource(
    database=DuckLake(
        metadata="postgres:dbname=lake host=pg",
        data_path="s3://bucket/lake/",
        options={"expire_older_than": "7 days"},
    )
)

Attributes: