dbt Adapter (`dbt-benostreamdb`)¶

The dbt-benostreamdb adapter enables analytics engineering, automated data transformations, and data lineage directly over BenoStreamDB and Apache Iceberg tables.

Overview¶

With dbt-benostreamdb, you can define models using familiar dbt SQL workflows while taking full advantage of BenoStreamDB’s native capabilities:

Iceberg Table & View Materializations: Build standard Parquet tables and metadata views without moving data out of your lakehouse.
Incremental Materializations: Efficient append and merge models leveraging primary keys and deletion vectors.
Vector & Hybrid Search Macros: Embedded macros for vector distance scoring and hybrid retrieval (benostreamdb.vector_search(...)).
Graph Traversal Macros: Native relationship walks (benostreamdb.graph_neighbors(...), benostreamdb.graph_shortest_path(...)) over edge tables.

Installation¶

Install the adapter via pip:

pip install dbt-benostreamdb

Or install from source within the repository:

cd dbt-benostreamdb
pip install -e .

Connection Profile (`profiles.yml`)¶

Configure your target connection in ~/.dbt/profiles.yml:

benostreamdb_profile:
  target: dev
  outputs:
    dev:
      type: benostreamdb
      database: default
      schema: public
      threads: 4
      uri: "file:///path/to/benostreamdb/warehouse"

For cloud object storage targets (S3 / GCS / Azure), specify the storage URI:

benostreamdb_profile:
  target: prod
  outputs:
    prod:
      type: benostreamdb
      database: default
      schema: analytics
      threads: 8
      uri: "s3://my-lakehouse-bucket/warehouse"

Remote runs: `dbt-trino` (no warehouse mount)¶

dbt-benostreamdb is an embedded adapter — it opens the warehouse in-process, so the path/URI must be reachable from wherever dbt runs. For Airflow, Cloud Run, or CI, point dbt-trino at a Trino coordinator that has the trino-benostreamdb plugin installed:

benostreamdb_remote:
  target: prod
  outputs:
    prod:
      type: trino
      method: none          # or ldap / jwt, per your coordinator
      host: trino.internal
      port: 8080
      catalog: benostream   # must match connector.name=benostream in etc/catalog
      schema: default
      threads: 8

This is the supported remote path: the connector already pushes system.vector_search, system.hybrid_search (including its filter argument), system.list_indexes, and system.regional_drift_search down to the engine, so remote dbt models keep the same search semantics as embedded runs. Keep the embedded adapter for local development and for Python-API features such as index building.

Which one to use¶

	`dbt-benostreamdb` (embedded)	`dbt-trino` + `trino-benostreamdb`
Infrastructure	none — `pip install` + a warehouse path	Trino coordinator + JVM + plugin + catalog file
Where the warehouse must live	reachable from the dbt process (mounted path / object store creds)	coordinator only
Concurrency / auth / isolation	single process	Trino’s (users, resource groups, queues)
Search functions	Python API and SQL macros	`system.*` table functions, pushed down
Python-API features (index builds, embeddings, graph helpers)	yes	no
Best for	laptops, CI, single-node lakehouse, notebooks	Airflow, Cloud Run, multi-team clusters

The embedded adapter exists precisely so that “run dbt against BenoStreamDB” does not require standing up a Trino cluster — that is the dependency it removes, and it is why both paths are kept: the engine is the same, only the transport differs. Keep the adapter thin (a SQL adapter over the engine’s session) and do not grow transport-only features in it; anything that needs a cluster belongs on the Trino path.

Type mapping¶

datetime columns map to timestamp (previously this raised NotImplementedException), next to the existing text → varchar, number → double, boolean → boolean, date → date, and time → time mappings. Trino exposes the same columns as TIMESTAMP, so a model built embedded also runs unchanged through dbt-trino.

Supported Materializations¶

1. Table Materialization¶

Creates an authoritative Apache Iceberg table:

{{ config(
    materialized='table',
    format_version=2
) }}

SELECT
    id,
    title,
    category,
    embedding
FROM {{ ref('stg_documents') }}
WHERE is_active = true

2. Incremental Materialization¶

Processes only new or updated rows since the last dbt execution:

{{ config(
    materialized='incremental',
    unique_key='id'
) }}

SELECT
    id,
    user_id,
    event_name,
    event_timestamp
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
    WHERE event_timestamp > (SELECT max(event_timestamp) FROM {{ this }})
{% endif %}

Vector & Search Macros¶

BenoStreamDB exposes helper macros for similarity queries and vector analytics:

{{ config(materialized='table') }}

SELECT
    id,
    title,
    embedding <-> '[0.12, 0.45, -0.23]'::vector AS l2_dist,
    embedding <=> '[0.12, 0.45, -0.23]'::vector AS cosine_dist
FROM {{ ref('catalog_items') }}
ORDER BY cosine_dist ASC
LIMIT 50

Graph Traversal Macros¶

Query relationship models and edge tables directly in your dbt models:

{{ config(materialized='table') }}

-- Traverse 2 hops from seed entity 101
SELECT *
FROM graph_neighbors('{{ ref("stg_edges") }}', '101', 2, 'auto')

Table-Level Cluster Indexing & Warmup Macros¶

For high-scale vector models (multi-million rows), you can register post-hooks in your model configuration to trigger global Voronoi cluster index builds and cache warming immediately following table materializations:

{{ config(
    materialized='table',
    post_hook=[
        rebuild_cluster_index(this, column='embedding', k=1024, m=16, ef_construction=200, target_shard_mb=512),
        warmup(this, n_probes=500, max_memory_gb=8.0)
    ]
) }}

SELECT
    id,
    title,
    embedding
FROM {{ ref('raw_embeddings') }}

Macros available in macros/ddl.sql:

{{ rebuild_cluster_index(relation, column='embedding', k=1024, m=16, ef_construction=200, target_shard_mb=512) }}: Emits ALTER TABLE <relation> EXECUTE rebuild_cluster_index(...), consolidating multi-segment vectors into rolling $\le 512\text{ MiB}$ Puffin index shards sealed into the Iceberg manifest.
{{ warmup(relation, n_probes=100, max_memory_gb=8.0) }}: Emits ALTER TABLE <relation> EXECUTE warmup(...), probing the Voronoi centroid router to prefault index shards into the OS page cache before live queries arrive.

Testing & Verification¶

Run standard dbt commands to validate and execute models:

# Test connection
dbt debug

# Run models
dbt run

# Run tests
dbt test

dbt Adapter (dbt-benostreamdb)¶