Changes not visible to new connections when calling dbtRunner() multiple times in same process with DuckDB

[rabo-unumed]

Current Behavior

When calling dbtRunner().invoke() multiple times in the same Python process with different --vars, changes from subsequent runs are not visible to new DuckDB connections. The persistent _ENV connection sees the correct (new) data, but fresh connections see stale data from a previous run.

This causes issues in batch processing scenarios where the same model is run repeatedly with different parameters.

Expected Behavior

I would expect the dbtRunner to give me the same outcome as if I called the CLI several times manually.

Steps To Reproduce

Was a bit stumped by this, and another colleague had been able to run the script many times in the past without the issue occurring, whereas I got it consistently. (an important note might be that he was running on Windows whereas I was on WSL2)
I left an agent running for a long time trying to find the cause, and it did manage to set this MWE up which seems to trigger the bug on my environment.

Minimal Working Example

Directory Structure

mwe_project/
├── dbt_project.yml
├── profiles.yml
├── models/
│   └── test_model.sql
└── test_bug.py

dbt_project.yml

name: mwe_project
version: '1.0.0'
profile: mwe_project
model-paths: ['models']

profiles.yml

mwe_project:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: 'test.duckdb'

models/test_model.sql

{{ config(materialized="table") }}
select '{{ var("my_var", "default") }}' as label

test_bug.py

"""
Reproduces dbt-duckdb bug: changes not visible to new connections
when calling dbtRunner() multiple times in same process.
"""
import os
import shutil
from pathlib import Path

import duckdb
from dbt.adapters.duckdb.connections import DuckDBConnectionManager
from dbt.cli.main import dbtRunner

PROJECT_DIR = Path(__file__).parent
os.chdir(PROJECT_DIR)

DB_PATH = PROJECT_DIR / "test.duckdb"

# Clean start
for f in [DB_PATH, PROJECT_DIR / "target"]:
    if f.exists():
        if f.is_dir():
            shutil.rmtree(f)
        else:
            f.unlink()

print(f"Database: {DB_PATH}")
print("=" * 60)

# Clear any previous _ENV state
DuckDBConnectionManager.close_all_connections()

values = ["VALUE_A", "VALUE_B"]

for val in values:
    print(f"\n>>> Running dbt with my_var={val}")

    runner = dbtRunner()
    result = runner.invoke([
        "run",
        "--select", "test_model",
        "--vars", f'{{"my_var": "{val}"}}',
        "--profiles-dir", str(PROJECT_DIR),
        "--project-dir", str(PROJECT_DIR),
    ])

    print(f"    dbt result: {'SUCCESS' if result.success else 'FAILED'}")

    # Check what _ENV connection sees
    env_val = None
    if DuckDBConnectionManager._ENV and DuckDBConnectionManager._ENV.conn:
        cursor = DuckDBConnectionManager._ENV.conn.cursor()
        env_val = cursor.execute("SELECT label FROM test_model LIMIT 1").fetchone()[0]
        cursor.close()
    print(f"    _ENV sees: {env_val}")

    # Check what a fresh connection sees
    conn = duckdb.connect(str(DB_PATH))
    fresh_val = conn.execute("SELECT label FROM test_model LIMIT 1").fetchone()[0]
    conn.close()
    print(f"    Fresh connection sees: {fresh_val}")

    if env_val != fresh_val:
        print(f"    *** BUG! _ENV has '{env_val}' but fresh sees stale '{fresh_val}' ***")
    else:
        print(f"    OK: Both see same value")

print("\n" + "=" * 60)
print("If bug present: Second run shows _ENV=VALUE_B but Fresh=VALUE_A")

Expected Output (with bug)

Database: /path/to/mwe_project/test.duckdb
============================================================

>>> Running dbt with my_var=VALUE_A
    dbt result: SUCCESS
    _ENV sees: VALUE_A
    Fresh connection sees: VALUE_A
    OK: Both see same value

>>> Running dbt with my_var=VALUE_B
    dbt result: SUCCESS
    _ENV sees: VALUE_B
    Fresh connection sees: VALUE_A
    *** BUG! _ENV has 'VALUE_B' but fresh sees stale 'VALUE_A' ***

============================================================
If bug present: Second run shows _ENV=VALUE_B but Fresh=VALUE_A

Environment

- **dbt-core**: 1.11.3
- **dbt-duckdb**: 1.10.0
- **duckdb**: 1.4.4
- **Python**: 3.10.12
- **OS**: Linux (through WSL2 on Windows 11)

Additional Context

It might be somewhat related to this ticket #616.

Original Use Case (Pseudocode)

We encountered this bug when generating reports for multiple entities in a loop:

from dbt.cli.main import dbtRunner
import duckdb

DB_PATH = "path/to/database.duckdb"

# Get list of entities to process
conn = duckdb.connect(DB_PATH)
entities = conn.execute("SELECT id FROM entities ORDER BY id").fetchall()
conn.close()

# Process each entity
for entity_id in entities:
    # Run dbt model with entity-specific vars
    vars = f'{{"entity_id": "{entity_id}", "report_period": "2025-02"}}'

    runner = dbtRunner()
    runner.invoke(["run", "--select", "entity_report", "--vars", vars])

    # Read results - THIS RETURNS STALE DATA FROM FIRST ITERATION
    conn = duckdb.connect(DB_PATH)
    df = conn.execute("SELECT * FROM entity_report").fetchdf()
    conn.close()

    # df contains data from the FIRST entity, not the current one!
    save_report(df, entity_id)

Expected: Each iteration processes and returns data for the current entity.
Actual: All iterations return data from the first entity processed.

Root Cause (as inferred by agent)

The DuckDBConnectionManager._ENV singleton persists across dbtRunner() invocations. Changes made in subsequent runs appear to not be committed/flushed to disk, so new connections see the previously committed state.

In connections.py:

@classmethod
def open(cls, connection: Connection) -> Connection:
    # ...
    with cls._LOCK:
        if not cls._ENV or cls._ENV.creds != credentials:
            cls._ENV = environments.create(credentials)  # Only creates new env if creds differ
        connection.handle = cls._ENV.handle()

Workaround

Call DuckDBConnectionManager.close_all_connections() after each dbtRunner().invoke():

from dbt.adapters.duckdb.connections import DuckDBConnectionManager

for entity_id in entities:
    runner = dbtRunner()
    runner.invoke(["run", "--select", "entity_report", "--vars", vars])

    # WORKAROUND: Force connection cleanup
    DuckDBConnectionManager.close_all_connections()

    # Now fresh connections see correct data
    conn = duckdb.connect(DB_PATH)
    df = conn.execute("SELECT * FROM entity_report").fetchdf()
    # ...