From 1f2ac7b78d5eed01ba728602bf0fcf8e9d9a01ff Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sun, 20 Sep 2026 11:58:02 +0900
Subject: [PATCH 1/6] fix(duckdb): use regular tables instead of temp tables
 under dbt Fusion

---
 .../utils/table_operations/create_table_as.sql | 13 +++++++++++--
 .../has_temp_table_support.sql                 | 10 ++++++++++
 .../table_operations/make_temp_relation.sql    | 18 ++++++++++++++++++
 3 files changed, 39 insertions(+), 2 deletions(-)

diff --git a/macros/utils/table_operations/create_table_as.sql b/macros/utils/table_operations/create_table_as.sql
index b56087bde..189fbd084 100644
--- a/macros/utils/table_operations/create_table_as.sql
+++ b/macros/utils/table_operations/create_table_as.sql
@@ -114,8 +114,17 @@
 {% macro duckdb__edr_get_create_table_as_sql(
     temporary, relation, sql_query, expiration_hours=none
 ) %}
-  create or replace {% if temporary %} temporary {% endif %} table {{ relation }}
-  as {{ sql_query }}
+    {% if temporary and elementary.is_dbt_fusion() %}
+        {# dbt-fusion uses connection pooling - DuckDB temp tables are session-scoped
+       and aren't visible in other sessions.  DuckDB also rejects TEMPORARY on a
+       qualified name.  Create regular tables instead - these are cleaned up by
+       Elementary's normal cleanup logic. #}
+    create or replace table {{ relation }}
+    as {{ sql_query }}
+    {% else %}
+    create or replace {% if temporary %} temporary {% endif %} table {{ relation }}
+    as {{ sql_query }}
+    {% endif %}
 {% endmacro %}
 
 {% macro trino__edr_get_create_table_as_sql(
diff --git a/macros/utils/table_operations/has_temp_table_support.sql b/macros/utils/table_operations/has_temp_table_support.sql
index 988267675..6b311dd7d 100644
--- a/macros/utils/table_operations/has_temp_table_support.sql
+++ b/macros/utils/table_operations/has_temp_table_support.sql
@@ -33,3 +33,13 @@
     {% else %} {% do return(true) %}
     {% endif %}
 {% endmacro %}
+
+
+{% macro duckdb__has_temp_table_support() %}
+    {# DuckDB temp tables are session-scoped and dbt-fusion runs each statement on a
+       pooled connection, so they are not visible to the next statement.
+       Use regular tables with cleanup instead. #}
+    {% if elementary.is_dbt_fusion() %} {% do return(false) %}
+    {% else %} {% do return(true) %}
+    {% endif %}
+{% endmacro %}
diff --git a/macros/utils/table_operations/make_temp_relation.sql b/macros/utils/table_operations/make_temp_relation.sql
index bcf031937..c28cc404d 100644
--- a/macros/utils/table_operations/make_temp_relation.sql
+++ b/macros/utils/table_operations/make_temp_relation.sql
@@ -90,6 +90,24 @@
     {% endif %}
 {% endmacro %}
 
+{% macro duckdb__edr_make_temp_relation(base_relation, suffix) %}
+    {% if elementary.is_dbt_fusion() %}
+        {# Workaround for dbt-fusion connection pooling - create regular relations
+           with explicit schema/database instead of temp relations #}
+        {% set tmp_identifier = elementary.table_name_with_suffix(
+            base_relation.identifier, suffix
+        ) %}
+        {% set tmp_relation = api.Relation.create(
+            identifier=tmp_identifier,
+            schema=base_relation.schema,
+            database=base_relation.database,
+            type="table",
+        ) %}
+        {% do return(tmp_relation) %}
+    {% else %} {% do return(dbt.make_temp_relation(base_relation, suffix)) %}
+    {% endif %}
+{% endmacro %}
+
 {% macro databricks__edr_make_temp_relation(base_relation, suffix) %}
     {% set tmp_identifier = elementary.table_name_with_suffix(
         base_relation.identifier, suffix

From 65d35be082cce38e8d5d5a3dc28fa61f399fe603 Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sun, 20 Sep 2026 11:58:26 +0900
Subject: [PATCH 2/6] test(duckdb): allow the integration tests to run against
 dbt Fusion

---
 .../tests/adapter_query_runner.py             | 26 +++++++++++++++++++
 integration_tests/tests/conftest.py           |  1 +
 2 files changed, 27 insertions(+)

diff --git a/integration_tests/tests/adapter_query_runner.py b/integration_tests/tests/adapter_query_runner.py
index c45cbb83a..90e25987f 100644
--- a/integration_tests/tests/adapter_query_runner.py
+++ b/integration_tests/tests/adapter_query_runner.py
@@ -248,6 +248,31 @@ def execute_sql(self, sql: str) -> None:
         """Execute a SQL statement that does not return results (DDL/DML)."""
         with self._adapter.connection_named("execute_sql"):
             self._adapter.execute(sql, fetch=False)
+        self._release_connections()
+
+    def _release_connections(self) -> None:
+        """Release the adapter connection after every query.
+
+        DuckDB is embedded and takes an exclusive lock on the database file, so an
+        idle connection held here blocks any dbt run that happens in a separate
+        process (`--runner-method fusion` / `subprocess`).  ``cleanup_all`` alone is
+        not enough: ``DuckDBConnectionManager._ENV`` caches the underlying
+        ``duckdb`` connection at class level and only closes it once its handle
+        count drops to zero, so the file stays locked.  Drop the cached environment
+        explicitly; the next query recreates it.
+
+        No-op for client/server warehouses, which allow concurrent connections.
+        """
+        self._adapter.connections.cleanup_all()
+        if self._adapter.type() != "duckdb":
+            return
+
+        from dbt.adapters.duckdb.connections import DuckDBConnectionManager
+
+        with DuckDBConnectionManager._LOCK:
+            if DuckDBConnectionManager._ENV is not None:
+                DuckDBConnectionManager._ENV.close()
+                DuckDBConnectionManager._ENV = None
 
     @property
     def schema_name(self) -> str:
@@ -269,6 +294,7 @@ def run_query(self, prerendered_query: str) -> List[Dict[str, Any]]:
         sql = self.resolve_refs(prerendered_query)
         with self._adapter.connection_named("run_query"):
             _response, table = self._adapter.execute(sql, fetch=True)
+        self._release_connections()
 
         # Convert agate Table → list[dict] matching agate_to_dicts behaviour
         columns = [c.lower() for c in table.column_names]
diff --git a/integration_tests/tests/conftest.py b/integration_tests/tests/conftest.py
index 579c5f70e..10b1dacda 100644
--- a/integration_tests/tests/conftest.py
+++ b/integration_tests/tests/conftest.py
@@ -18,6 +18,7 @@
     "bigquery",
     "redshift",
     "databricks_catalog",
+    "duckdb",
 ]
 
 logger = get_logger(__name__)

From 02378beaf0c89e9bddb6f32deb75079728ceb947 Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sun, 20 Sep 2026 14:12:38 +0900
Subject: [PATCH 3/6] test(duckdb): only release the DuckDB file lock, and
 release it from finally

---
 .../tests/adapter_query_runner.py             | 54 +++++++++++--------
 1 file changed, 33 insertions(+), 21 deletions(-)

diff --git a/integration_tests/tests/adapter_query_runner.py b/integration_tests/tests/adapter_query_runner.py
index 90e25987f..e6d1c3d39 100644
--- a/integration_tests/tests/adapter_query_runner.py
+++ b/integration_tests/tests/adapter_query_runner.py
@@ -246,32 +246,42 @@ def has_non_ref_jinja(query: str) -> bool:
 
     def execute_sql(self, sql: str) -> None:
         """Execute a SQL statement that does not return results (DDL/DML)."""
-        with self._adapter.connection_named("execute_sql"):
-            self._adapter.execute(sql, fetch=False)
-        self._release_connections()
-
-    def _release_connections(self) -> None:
-        """Release the adapter connection after every query.
-
-        DuckDB is embedded and takes an exclusive lock on the database file, so an
-        idle connection held here blocks any dbt run that happens in a separate
-        process (`--runner-method fusion` / `subprocess`).  ``cleanup_all`` alone is
-        not enough: ``DuckDBConnectionManager._ENV`` caches the underlying
-        ``duckdb`` connection at class level and only closes it once its handle
-        count drops to zero, so the file stays locked.  Drop the cached environment
-        explicitly; the next query recreates it.
-
-        No-op for client/server warehouses, which allow concurrent connections.
+        try:
+            with self._adapter.connection_named("execute_sql"):
+                self._adapter.execute(sql, fetch=False)
+        finally:
+            self._release_duckdb_file_lock()
+
+    def _release_duckdb_file_lock(self) -> None:
+        """Release the DuckDB connection so an out-of-process dbt run can take the file lock.
+
+        A file-backed DuckDB database allows a single read-write process at a time,
+        so the connection cached in ``DuckDBConnectionManager._ENV`` must be closed
+        before dbt runs in another process (`--runner-method fusion` / `subprocess`).
+        The next query recreates it.
+
+        No-op for non-DuckDB adapters, ``path: ":memory:"`` (closing it would discard
+        the database) and MotherDuck (hosted, not a local file).
         """
-        self._adapter.connections.cleanup_all()
         if self._adapter.type() != "duckdb":
             return
 
+        credentials = self._adapter.config.credentials
+        if credentials.path == ":memory:" or credentials.is_motherduck:
+            return
+
         from dbt.adapters.duckdb.connections import DuckDBConnectionManager
 
+        self._adapter.connections.cleanup_all()
         with DuckDBConnectionManager._LOCK:
             if DuckDBConnectionManager._ENV is not None:
-                DuckDBConnectionManager._ENV.close()
+                try:
+                    DuckDBConnectionManager._ENV.close()
+                except Exception as exc:
+                    # Called from a `finally`, so never mask the original error.
+                    logger.warning(
+                        f"Ignoring error while closing the DuckDB environment: {exc}"
+                    )
                 DuckDBConnectionManager._ENV = None
 
     @property
@@ -292,9 +302,11 @@ def run_query(self, prerendered_query: str) -> List[Dict[str, Any]]:
         if self.has_non_ref_jinja(prerendered_query):
             raise UnsupportedJinjaError(prerendered_query)
         sql = self.resolve_refs(prerendered_query)
-        with self._adapter.connection_named("run_query"):
-            _response, table = self._adapter.execute(sql, fetch=True)
-        self._release_connections()
+        try:
+            with self._adapter.connection_named("run_query"):
+                _response, table = self._adapter.execute(sql, fetch=True)
+        finally:
+            self._release_duckdb_file_lock()
 
         # Convert agate Table → list[dict] matching agate_to_dicts behaviour
         columns = [c.lower() for c in table.column_names]

From 0f8ac4f16a8c46388ec648e785ea18e5f92c243e Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sat, 26 Sep 2026 19:17:49 +0900
Subject: [PATCH 4/6] fix(duckdb): commit the temp table drop and register the
 test result relation

Review follow-ups on the Fusion path:

- handle_tests_results.sql: commit after fully_drop_relation at both drop sites,
  so the drop survives dbt's post-on-run-end ROLLBACK. Gated on DuckDB like the
  existing commit five lines above.
- test.sql: register the relation from create_test_result_temp_table in
  temp_test_table_relations_map, so the on-run-end cleanup reaches it.
- create_table_as.sql: replace the duplicated duckdb__ body with
  `set temporary = temporary and not is_dbt_fusion()`; the SQL is byte-identical
  to upstream again.
- make_temp_relation.sql / has_temp_table_support.sql: redshift__ and duckdb__
  share elementary._fusion_regular_temp_relation(), and the predicate is a
  one-liner for both. Behaviour unchanged for Redshift.
---
 macros/edr/materializations/test/test.sql     |  9 ++++
 .../tests/on_run_end/handle_tests_results.sql |  4 ++
 .../table_operations/create_table_as.sql      | 18 +++-----
 .../has_temp_table_support.sql                | 18 +++-----
 .../table_operations/make_temp_relation.sql   | 43 ++++++++++---------
 5 files changed, 48 insertions(+), 44 deletions(-)

diff --git a/macros/edr/materializations/test/test.sql b/macros/edr/materializations/test/test.sql
index 90bb23487..d4aba5822 100644
--- a/macros/edr/materializations/test/test.sql
+++ b/macros/edr/materializations/test/test.sql
@@ -193,6 +193,15 @@
     {% set database, schema = elementary.get_package_database_and_schema() %}
     {% set test_id = model["alias"] %}
     {% set relation = elementary.create_temp_table(database, schema, test_id, sql) %}
+    {# On DuckDB under dbt-fusion this is a regular table rather than a temp one, so
+       register it for the on_run_end cleanup - otherwise it survives until the stale
+       cleanup runs. #}
+    {% if target.type == "duckdb" and elementary.is_dbt_fusion() %}
+        {% set test_entry = elementary.get_cache(
+            "temp_test_table_relations_map"
+        ).setdefault(test_id, {}) %}
+        {% do test_entry.update({"test_result": relation}) %}
+    {% endif %}
     {% set new_sql %}
     select * from {{ relation }}
     {% endset %}
diff --git a/macros/edr/tests/on_run_end/handle_tests_results.sql b/macros/edr/tests/on_run_end/handle_tests_results.sql
index b41114080..d7e33a8e6 100644
--- a/macros/edr/tests/on_run_end/handle_tests_results.sql
+++ b/macros/edr/tests/on_run_end/handle_tests_results.sql
@@ -227,6 +227,8 @@
 
     {% if not elementary.has_temp_table_support() %}
         {% do elementary.fully_drop_relation(temp_relation) %}
+        {# DuckDB: commit the drop too, so it survives the same ROLLBACK #}
+        {% if target.type == "duckdb" %} {% do adapter.commit() %} {% endif %}
     {% endif %}
 {% endmacro %}
 
@@ -287,6 +289,8 @@
 
     {% if not elementary.has_temp_table_support() %}
         {% do elementary.fully_drop_relation(temp_relation) %}
+        {# DuckDB: commit the drop too, so it survives the same ROLLBACK #}
+        {% if target.type == "duckdb" %} {% do adapter.commit() %} {% endif %}
     {% endif %}
 {% endmacro %}
 
diff --git a/macros/utils/table_operations/create_table_as.sql b/macros/utils/table_operations/create_table_as.sql
index e15d916bb..9db9788ab 100644
--- a/macros/utils/table_operations/create_table_as.sql
+++ b/macros/utils/table_operations/create_table_as.sql
@@ -120,17 +120,13 @@
 {% macro duckdb__edr_get_create_table_as_sql(
     temporary, relation, sql_query, expiration_hours=none
 ) %}
-    {% if temporary and elementary.is_dbt_fusion() %}
-        {# dbt-fusion uses connection pooling - DuckDB temp tables are session-scoped
-       and aren't visible in other sessions.  DuckDB also rejects TEMPORARY on a
-       qualified name.  Create regular tables instead - these are cleaned up by
-       Elementary's normal cleanup logic. #}
-    create or replace table {{ relation }}
-    as {{ sql_query }}
-    {% else %}
-    create or replace {% if temporary %} temporary {% endif %} table {{ relation }}
-    as {{ sql_query }}
-    {% endif %}
+    {# dbt-fusion uses connection pooling - DuckDB temp tables are session-scoped and
+       aren't visible in other sessions.  DuckDB also rejects TEMPORARY on a qualified
+       name.  Create regular tables instead - these are cleaned up by Elementary's
+       normal cleanup logic. #}
+    {% set temporary = temporary and not elementary.is_dbt_fusion() %}
+  create or replace {% if temporary %} temporary {% endif %} table {{ relation }}
+  as {{ sql_query }}
 {% endmacro %}
 
 {% macro trino__edr_get_create_table_as_sql(
diff --git a/macros/utils/table_operations/has_temp_table_support.sql b/macros/utils/table_operations/has_temp_table_support.sql
index 6b311dd7d..3af7cf0d7 100644
--- a/macros/utils/table_operations/has_temp_table_support.sql
+++ b/macros/utils/table_operations/has_temp_table_support.sql
@@ -25,21 +25,13 @@
 {% endmacro %}
 
 
+{# Session-scoped temp tables are invisible to the next statement under dbt-fusion,
+   which runs each statement on a pooled connection ("relation does not exist").
+   These adapters fall back to regular tables with cleanup instead. #}
 {% macro redshift__has_temp_table_support() %}
-    {# dbt-fusion uses connection pooling, so temp tables created in one session
-       are not visible in other sessions, causing "relation does not exist" errors.
-       Use regular tables with cleanup instead. #}
-    {% if elementary.is_dbt_fusion() %} {% do return(false) %}
-    {% else %} {% do return(true) %}
-    {% endif %}
+    {% do return(not elementary.is_dbt_fusion()) %}
 {% endmacro %}
 
-
 {% macro duckdb__has_temp_table_support() %}
-    {# DuckDB temp tables are session-scoped and dbt-fusion runs each statement on a
-       pooled connection, so they are not visible to the next statement.
-       Use regular tables with cleanup instead. #}
-    {% if elementary.is_dbt_fusion() %} {% do return(false) %}
-    {% else %} {% do return(true) %}
-    {% endif %}
+    {% do return(not elementary.is_dbt_fusion()) %}
 {% endmacro %}
diff --git a/macros/utils/table_operations/make_temp_relation.sql b/macros/utils/table_operations/make_temp_relation.sql
index c28cc404d..14050eff0 100644
--- a/macros/utils/table_operations/make_temp_relation.sql
+++ b/macros/utils/table_operations/make_temp_relation.sql
@@ -72,38 +72,41 @@
     {{ return(elementary.spark__edr_make_temp_relation(base_relation, suffix)) }}
 {% endmacro %}
 
-{% macro redshift__edr_make_temp_relation(base_relation, suffix) %}
-    {% if elementary.is_dbt_fusion() %}
-        {# Workaround for dbt-fusion temp table metadata bug - create regular relations
-           with explicit schema/database instead of temp relations #}
-        {% set tmp_identifier = elementary.table_name_with_suffix(
-            base_relation.identifier, suffix
-        ) %}
-        {% set tmp_relation = api.Relation.create(
+{% macro _fusion_regular_temp_relation(base_relation, suffix) %}
+    {# dbt-fusion runs each statement on a pooled connection, so a session-scoped temp
+       relation is not visible to the statement that reads it.  Build a regular relation
+       with an explicit database/schema instead - Elementary's normal cleanup drops it. #}
+    {% set tmp_identifier = elementary.table_name_with_suffix(
+        base_relation.identifier, suffix
+    ) %}
+    {% do return(
+        api.Relation.create(
             identifier=tmp_identifier,
             schema=base_relation.schema,
             database=base_relation.database,
             type="table",
+        )
+    ) %}
+{% endmacro %}
+
+{% macro redshift__edr_make_temp_relation(base_relation, suffix) %}
+    {% if elementary.is_dbt_fusion() %}
+        {# Workaround for dbt-fusion temp table metadata bug - create regular relations
+           with explicit schema/database instead of temp relations #}
+        {% do return(
+            elementary._fusion_regular_temp_relation(base_relation, suffix)
         ) %}
-        {% do return(tmp_relation) %}
     {% else %} {% do return(dbt.make_temp_relation(base_relation, suffix)) %}
     {% endif %}
 {% endmacro %}
 
 {% macro duckdb__edr_make_temp_relation(base_relation, suffix) %}
     {% if elementary.is_dbt_fusion() %}
-        {# Workaround for dbt-fusion connection pooling - create regular relations
-           with explicit schema/database instead of temp relations #}
-        {% set tmp_identifier = elementary.table_name_with_suffix(
-            base_relation.identifier, suffix
+        {# DuckDB additionally rejects TEMPORARY on a qualified name, so the regular
+           relation is required rather than merely convenient. #}
+        {% do return(
+            elementary._fusion_regular_temp_relation(base_relation, suffix)
         ) %}
-        {% set tmp_relation = api.Relation.create(
-            identifier=tmp_identifier,
-            schema=base_relation.schema,
-            database=base_relation.database,
-            type="table",
-        ) %}
-        {% do return(tmp_relation) %}
     {% else %} {% do return(dbt.make_temp_relation(base_relation, suffix)) %}
     {% endif %}
 {% endmacro %}

From f7256eabdd56dd8bc61754f8f93323bbf748af02 Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sat, 26 Sep 2026 19:18:30 +0900
Subject: [PATCH 5/6] test(duckdb): release the lock via the public API and
 only when dbt shells out

- Use DuckDBConnectionManager.close_all_connections() instead of touching
  _ENV/_LOCK. Dropping the last reference closes the connection via
  LocalEnvironment.__del__, which is enough to hand over the file lock.
- Wrap both release steps in one try; a failure degrades to a warning so nothing
  reached from the finally can mask the original query error.
- Skip the release entirely when dbt runs in-process (RunnerMethod.API).
- Fail fast when the duckdb profile is :memory: and a dbt 2.0 runner is
  requested, instead of letting the whole suite fail on the lock.
- Replace the four RunnerMethod.FUSION comparisons with a DBT2_RUNNERS tuple, so
  --runner-method dbt2 stops being a silent no-op.
---
 .../tests/adapter_query_runner.py             | 51 ++++++++++++-------
 integration_tests/tests/conftest.py           | 41 ++++++++++++---
 integration_tests/tests/dbt_project.py        |  7 ++-
 3 files changed, 73 insertions(+), 26 deletions(-)

diff --git a/integration_tests/tests/adapter_query_runner.py b/integration_tests/tests/adapter_query_runner.py
index e6d1c3d39..fe8e75b56 100644
--- a/integration_tests/tests/adapter_query_runner.py
+++ b/integration_tests/tests/adapter_query_runner.py
@@ -15,6 +15,7 @@
 from typing import Any, Dict, List, Optional
 
 from dbt.adapters.base import BaseAdapter
+from elementary.clients.dbt.factory import RunnerMethod, get_dbt_runner_method
 from logger import get_logger
 
 logger = get_logger(__name__)
@@ -76,11 +77,23 @@ class AdapterQueryRunner:
         Path to the dbt project directory.
     target : str
         Name of the dbt target / profile output to use.
+    runner_method : RunnerMethod, optional
+        How the tests run dbt.  Only out-of-process runners need the DuckDB file
+        lock released between queries (see ``_release_duckdb_file_lock``).
     """
 
-    def __init__(self, project_dir: str, target: str) -> None:
+    def __init__(
+        self,
+        project_dir: str,
+        target: str,
+        runner_method: Optional[RunnerMethod] = None,
+    ) -> None:
         self._project_dir = project_dir
         self._target = target
+        # Every runner except API shells out, so it needs the file lock released.
+        self._dbt_runs_out_of_process = (
+            runner_method or get_dbt_runner_method()
+        ) != RunnerMethod.API
         self._adapter: BaseAdapter = self._create_adapter(project_dir, target)
         self._ref_map: Optional[Dict[str, str]] = None
         self._source_map: Optional[Dict[tuple, str]] = None
@@ -255,14 +268,19 @@ def execute_sql(self, sql: str) -> None:
     def _release_duckdb_file_lock(self) -> None:
         """Release the DuckDB connection so an out-of-process dbt run can take the file lock.
 
-        A file-backed DuckDB database allows a single read-write process at a time,
-        so the connection cached in ``DuckDBConnectionManager._ENV`` must be closed
-        before dbt runs in another process (`--runner-method fusion` / `subprocess`).
-        The next query recreates it.
+        A file-backed DuckDB database allows a single read-write process at a time, so
+        the environment cached on ``DuckDBConnectionManager`` has to be dropped before
+        dbt runs in another process (any ``--runner-method`` other than ``api``).  Dropping
+        the last reference closes the DuckDB connection via ``LocalEnvironment.__del__``;
+        the next query builds a new environment.
 
-        No-op for non-DuckDB adapters, ``path: ":memory:"`` (closing it would discard
-        the database) and MotherDuck (hosted, not a local file).
+        No-op unless dbt actually runs out of process, and for non-DuckDB adapters,
+        ``path: ":memory:"`` (closing it would discard the database) and MotherDuck
+        (hosted, not a local file).
         """
+        if not self._dbt_runs_out_of_process:
+            return
+
         if self._adapter.type() != "duckdb":
             return
 
@@ -272,17 +290,14 @@ def _release_duckdb_file_lock(self) -> None:
 
         from dbt.adapters.duckdb.connections import DuckDBConnectionManager
 
-        self._adapter.connections.cleanup_all()
-        with DuckDBConnectionManager._LOCK:
-            if DuckDBConnectionManager._ENV is not None:
-                try:
-                    DuckDBConnectionManager._ENV.close()
-                except Exception as exc:
-                    # Called from a `finally`, so never mask the original error.
-                    logger.warning(
-                        f"Ignoring error while closing the DuckDB environment: {exc}"
-                    )
-                DuckDBConnectionManager._ENV = None
+        # Called from a `finally`, so neither step may mask the original error.
+        try:
+            self._adapter.connections.cleanup_all()
+            DuckDBConnectionManager.close_all_connections()
+        except Exception as exc:
+            logger.warning(
+                f"Ignoring error while releasing the DuckDB connection: {exc}"
+            )
 
     @property
     def schema_name(self) -> str:
diff --git a/integration_tests/tests/conftest.py b/integration_tests/tests/conftest.py
index 10b1dacda..208e2991c 100644
--- a/integration_tests/tests/conftest.py
+++ b/integration_tests/tests/conftest.py
@@ -1,3 +1,4 @@
+import os
 import shutil
 from pathlib import Path
 from tempfile import mkdtemp
@@ -6,7 +7,12 @@
 import pytest
 import yaml
 from dbt.version import __version__ as dbt_version
-from dbt_project import PYTEST_XDIST_WORKER, SCHEMA_NAME_SUFFIX, DbtProject
+from dbt_project import (
+    DBT2_RUNNERS,
+    PYTEST_XDIST_WORKER,
+    SCHEMA_NAME_SUFFIX,
+    DbtProject,
+)
 from elementary.clients.dbt.factory import RunnerMethod
 from env import Environment
 from logger import get_logger
@@ -71,7 +77,7 @@ def _edit_packages_yml_to_include_absolute_elementary_package_path(
 def _remove_python_models_for_dbt_fusion(
     project_dir_copy: str, runner_method: Optional[RunnerMethod]
 ):
-    if runner_method != RunnerMethod.FUSION:
+    if runner_method not in DBT2_RUNNERS:
         return
 
     logger.info(f"Removing python tests for project {project_dir_copy}")
@@ -130,7 +136,7 @@ def only_on_targets(request, target: str):
 @pytest.fixture(autouse=True)
 def skip_for_dbt_fusion(request, runner_method: Optional[RunnerMethod]):
     if request.node.get_closest_marker("skip_for_dbt_fusion"):
-        if runner_method == RunnerMethod.FUSION:
+        if runner_method in DBT2_RUNNERS:
             pytest.skip("Test unsupported for dbt fusion")
 
 
@@ -170,16 +176,39 @@ def clear_on_end(request) -> bool:
     return request.config.getoption("--clear-on-end")
 
 
+def _profile_path(target: str) -> Optional[str]:
+    """Return the ``path`` configured for *target* in profiles.yml, if readable.
+
+    A profile that omits ``path`` gets dbt-duckdb's own default, so an implicit
+    in-memory database is reported as such.  Returns ``None`` when the profile
+    cannot be resolved, so callers only act on a value they positively read.
+    """
+    profiles_dir = os.environ.get("DBT_PROFILES_DIR", os.path.expanduser("~/.dbt"))
+    try:
+        project = yaml.safe_load((DBT_PROJECT_PATH / "dbt_project.yml").read_text())
+        profiles = yaml.safe_load((Path(profiles_dir) / "profiles.yml").read_text())
+        return profiles[project["profile"]]["outputs"][target].get("path", ":memory:")
+    except Exception:
+        return None
+
+
 @pytest.fixture(scope="session")
 def runner_method(request, target: str) -> Optional[RunnerMethod]:
     runner_method_str = request.config.getoption("--runner-method")
     if runner_method_str:
         runner_method = RunnerMethod(runner_method_str)
+        if runner_method in DBT2_RUNNERS and target not in DBT_FUSION_SUPPORTED_TARGETS:
+            raise ValueError(f"Fusion runner is not supported for target: {target}")
         if (
-            runner_method == RunnerMethod.FUSION
-            and target not in DBT_FUSION_SUPPORTED_TARGETS
+            runner_method in DBT2_RUNNERS
+            and target == "duckdb"
+            and _profile_path(target) == ":memory:"
         ):
-            raise ValueError(f"Fusion runner is not supported for target: {target}")
+            raise ValueError(
+                f"The {runner_method.value} runner runs dbt in a subprocess, which "
+                "cannot share an in-memory DuckDB database with the tests. Set "
+                "`path` to a file in the duckdb target of your profiles.yml."
+            )
         return runner_method
     return None
 
diff --git a/integration_tests/tests/dbt_project.py b/integration_tests/tests/dbt_project.py
index af32b9ca5..f499895c9 100644
--- a/integration_tests/tests/dbt_project.py
+++ b/integration_tests/tests/dbt_project.py
@@ -22,6 +22,9 @@
 PYTEST_XDIST_WORKER = os.environ.get("PYTEST_XDIST_WORKER", None)
 SCHEMA_NAME_SUFFIX = f"_{PYTEST_XDIST_WORKER}" if PYTEST_XDIST_WORKER else ""
 
+# Both spellings of the dbt 2.0 runner: FUSION is a legacy alias for DBT2.
+DBT2_RUNNERS = (RunnerMethod.DBT2, RunnerMethod.FUSION)
+
 _DEFAULT_VARS = {
     "disable_dbt_invocation_autoupload": True,
     "disable_dbt_artifacts_autoupload": True,
@@ -88,7 +91,7 @@ def _get_query_runner(self) -> AdapterQueryRunner:
         """Lazily initialize the direct adapter query runner."""
         if self._query_runner is None:
             self._query_runner = AdapterQueryRunner(
-                str(self.project_dir_path), self.target
+                str(self.project_dir_path), self.target, self.runner_method
             )
         return self._query_runner
 
@@ -395,7 +398,7 @@ def seed(self, data: List[dict], table_name: str):
     def _fix_seed_if_needed(self, table_name: str) -> None:
         # Hack for BigQuery - seems like we get empty strings instead of nulls in seeds, so we
         # fix them here.
-        if self.runner_method == RunnerMethod.FUSION and self.target == "bigquery":
+        if self.runner_method in DBT2_RUNNERS and self.target == "bigquery":
             self.dbt_runner.run_operation(
                 "elementary_tests.replace_empty_strings_with_nulls",
                 macro_args={"table_name": table_name},

From 736e38b1db20d7370628350ccc6a89f9bee1c31b Mon Sep 17 00:00:00 2001
From: junoha <3103358+junoha@users.noreply.github.com>
Date: Sat, 26 Sep 2026 19:56:05 +0900
Subject: [PATCH 6/6] test(duckdb): resolve the profile through dbt before the
 in-memory check

  _profile_path read profiles.yml with yaml.safe_load, so a `path` supplied as
  `{{ env_var(...) }}` was compared unrendered and an in-memory database slipped
  past the guard. Resolve the profile through dbt instead and read
  credentials.path, which renders env_var (hence set_invocation_context) and
  picks up dbt-duckdb's own `:memory:` default.
---
 integration_tests/tests/conftest.py | 35 +++++++++++++++++++++++------
 1 file changed, 28 insertions(+), 7 deletions(-)

diff --git a/integration_tests/tests/conftest.py b/integration_tests/tests/conftest.py
index 208e2991c..d36d83a85 100644
--- a/integration_tests/tests/conftest.py
+++ b/integration_tests/tests/conftest.py
@@ -177,17 +177,38 @@ def clear_on_end(request) -> bool:
 
 
 def _profile_path(target: str) -> Optional[str]:
-    """Return the ``path`` configured for *target* in profiles.yml, if readable.
+    """Return the ``path`` dbt resolves for *target*, if it can be resolved.
 
-    A profile that omits ``path`` gets dbt-duckdb's own default, so an implicit
-    in-memory database is reported as such.  Returns ``None`` when the profile
-    cannot be resolved, so callers only act on a value they positively read.
+    Goes through dbt's own profile handling instead of reading ``profiles.yml``
+    directly, so ``env_var`` and any other Jinja in ``path`` render the way dbt
+    will render it, and an omitted ``path`` picks up dbt-duckdb's ``:memory:``
+    default.  ``env_var`` needs the invocation context, hence the
+    ``set_invocation_context`` call.
+
+    Returns ``None`` when the profile cannot be resolved, so callers only act on
+    a value they positively read.
     """
+    from argparse import Namespace
+
+    from dbt.config.runtime import RuntimeConfig
+    from dbt.flags import set_from_args
+    from dbt_common.context import set_invocation_context
+
     profiles_dir = os.environ.get("DBT_PROFILES_DIR", os.path.expanduser("~/.dbt"))
+    args = Namespace(
+        project_dir=str(DBT_PROJECT_PATH),
+        profiles_dir=profiles_dir,
+        target=target,
+        threads=1,
+        vars={},
+        profile=None,
+        PROFILES_DIR=profiles_dir,
+        PROJECT_DIR=str(DBT_PROJECT_PATH),
+    )
     try:
-        project = yaml.safe_load((DBT_PROJECT_PATH / "dbt_project.yml").read_text())
-        profiles = yaml.safe_load((Path(profiles_dir) / "profiles.yml").read_text())
-        return profiles[project["profile"]]["outputs"][target].get("path", ":memory:")
+        set_invocation_context(os.environ)
+        set_from_args(args, None)
+        return getattr(RuntimeConfig.from_args(args).credentials, "path", None)
     except Exception:
         return None