# 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:

```bash
pip install dbt-benostreamdb
```

Or install from source within the repository:

```bash
cd dbt-benostreamdb
pip install -e .
```

---

## Connection Profile (`profiles.yml`)

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

```yaml
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:

```yaml
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:

```yaml
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:

```sql+jinja
{{ 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:

```sql+jinja
{{ 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:

```sql+jinja
{{ 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:

```sql+jinja
{{ 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:

```sql+jinja
{{ 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:

```bash
# Test connection
dbt debug

# Run models
dbt run

# Run tests
dbt test
```