Respect QUOTED_IDENTIFIERS_IGNORE_CASE with dynamic tables (#1115)

Co-authored-by: Colin Rogers <[email protected]>

Author: VersusFacit

dbt-snowflake/.changes/unreleased/Fixes-20250522-031531.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Respect QUOTED_IDENTIFIERS_IGNORE_CASE with dynamic tables.
+time: 2025-05-22T03:15:31.20101-07:00
+custom:
+    Author: versusfacit
+    Issue: "993"

dbt-snowflake/src/dbt/adapters/snowflake/impl.py

@@ -478,3 +478,60 @@ def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelat
             catalog_integration = self.get_catalog_integration(catalog)
             return catalog_integration.build_relation(model)
         return None
+
+    @available
+    def describe_dynamic_table(self, relation: RelationConfig) -> Dict[str, Any]:
+        """
+        Get all relevant metadata about a dynamic table to return as a dict to Agate Table row
+
+        Args:
+            relation (SnowflakeRelation): the relation to describe
+        """
+
+        original_val: Optional[str] = None
+        try:
+            # Store old QUOTED_IDENTIFIERS_IGNORE_CASE
+            show_param_sql = "show parameters like 'QUOTED_IDENTIFIERS_IGNORE_CASE' in SESSION"
+            show_param_res = self.execute(show_param_sql)
+            if show_param_res:
+                param_qid = show_param_res[0].query_id
+                scan_param_sql = f"SELECT * FROM TABLE(RESULT_SCAN('{param_qid}'))"
+                param_scan_res, rows = self.execute(scan_param_sql, fetch=True)
+                if param_scan_res and param_scan_res.code == "SUCCESS":
+                    try:
+                        original_val = rows[0][1]
+                    except (IndexError, TypeError):
+                        original_val = None
+
+            # falsify QUOTED_IDENTIFIERS_IGNORE_CASE for execution only
+            self.execute("alter session set QUOTED_IDENTIFIERS_IGNORE_CASE = FALSE", fetch=False)
+
+            show_sql = (
+                f"show dynamic tables like '{relation.identifier}' "
+                f"in schema {relation.database}.{relation.schema}"
+            )
+            show_res = self.execute(show_sql)
+            if show_res:
+                query_id = show_res[0].query_id
+                scan_sql = (
+                    "select "
+                    '   "name", "schema_name", "database_name", "text", "target_lag", "warehouse", '
+                    '   "refresh_mode" '
+                    f"from TABLE(RESULT_SCAN('{query_id}'))"
+                )
+                res, dt_table = self.execute(scan_sql, fetch=True)
+                if res.code != "SUCCESS":
+                    raise DbtRuntimeError(
+                        f"Could not get dynamic query metadata: {scan_sql} failed"
+                    )
+                return {"dynamic_table": dt_table}
+
+            return {"dynamic_table": None}
+        finally:
+            if original_val is None:
+                self.execute("ALTER SESSION UNSET QUOTED_IDENTIFIERS_IGNORE_CASE", fetch=False)
+            else:
+                bool_str = "TRUE" if original_val.strip().lower() == "true" else "FALSE"
+                self.execute(
+                    f"ALTER SESSION SET QUOTED_IDENTIFIERS_IGNORE_CASE = {bool_str}", fetch=False
+                )

dbt-snowflake/src/dbt/include/snowflake/macros/relations/dynamic_table/describe.sql

@@ -1,35 +1,4 @@
 {% macro snowflake__describe_dynamic_table(relation) %}
-{#-
-    Get all relevant metadata about a dynamic table
-
-    Args:
-    - relation: SnowflakeRelation - the relation to describe
-    Returns:
-        A dictionary with one or two entries depending on whether iceberg is enabled:
-        - dynamic_table: the metadata associated with an info schema dynamic table
--#}
-    {%- set _dynamic_table_sql -%}
-    alter session set quoted_identifiers_ignore_case = false;
-    show dynamic tables
-        like '{{ relation.identifier }}'
-        in schema {{ relation.database }}.{{ relation.schema }}
-    ;
-    select
-        "name",
-        "schema_name",
-        "database_name",
-        "text",
-        "target_lag",
-        "warehouse",
-        "refresh_mode"
-    from table(result_scan(last_query_id()))
-    ;
-    {%- endset -%}
-
-    {%- set results = {'dynamic_table': run_query(_dynamic_table_sql)} -%}
-
-    alter session unset quoted_identifiers_ignore_case;
-
+    {%- set results = adapter.describe_dynamic_table(relation) -%}
     {%- do return(results) -%}
-
 {% endmacro %}

dbt-snowflake/tests/functional/adapter/test_quoted_identifiers_flag.py

@@ -0,0 +1,84 @@
+import pytest
+from unittest.mock import MagicMock
+
+
+def _fake_relation(project, name="DUMMY_DT"):
+    """Return a bare-bones object with the three FQN fields."""
+    rel = MagicMock()
+    rel.identifier = name
+    rel.database = project.database
+    rel.schema = project.test_schema
+    return rel
+
+
+def _get_qiic_value(adapter):
+    """
+    Snowflake always yields one row; default is 'false'.
+    """
+    _, tbl = adapter.execute(
+        "SHOW PARAMETERS LIKE 'QUOTED_IDENTIFIERS_IGNORE_CASE' IN SESSION",
+        fetch=True,
+    )
+    # tbl.rows[0] is guaranteed by SHOW … IN SESSION
+    val = tbl.rows[0][1]
+    return val.strip().lower() if val else None  # e.g. 'true' or 'false'
+
+
+class TestQuotedIdentifiersFlag:
+    """
+    Ensure describe_dynamic_table() restores the session parameter
+    QUOTED_IDENTIFIERS_IGNORE_CASE to its original value.
+    """
+
+    @pytest.mark.parametrize("initial_setting", ["true", "false"])
+    def test_restores_parameter(self, project, initial_setting):
+        adapter = project.adapter
+
+        adapter.execute(
+            f"ALTER SESSION SET QUOTED_IDENTIFIERS_IGNORE_CASE = {initial_setting.upper()}",
+            fetch=False,
+        )
+        before = _get_qiic_value(adapter)
+        assert before == initial_setting
+
+        relation = _fake_relation(project)
+        adapter.describe_dynamic_table(relation)
+
+        after = _get_qiic_value(adapter)
+        assert after == before, f"Parameter not restored: before={before!r}, after={after!r}"
+
+
+class TestFlagRestoredOnSqlError:
+    """
+    Simulate adapter.execute raising an exception *after* the flag
+    has been flipped, and assert the flag is still reset.
+    """
+
+    @pytest.mark.parametrize("initial_setting", ["true", "false"])
+    def test_restored_on_exception(self, project, monkeypatch, initial_setting):
+        adapter = project.adapter
+
+        # initial flag state
+        adapter.execute(
+            f"ALTER SESSION SET QUOTED_IDENTIFIERS_IGNORE_CASE = {initial_setting.upper()}",
+            fetch=False,
+        )
+        before = _get_qiic_value(adapter)
+
+        # monkey-patch execute so the 5th call raises
+        call_count = {"n": 0}
+
+        def flaky_execute(sql, fetch=False):
+            call_count["n"] += 1
+            if call_count["n"] == 5:
+                raise RuntimeError("boom")
+            return adapter.__class__.execute(adapter, sql, fetch=fetch)
+
+        monkeypatch.setattr(adapter, "execute", flaky_execute)
+
+        relation = _fake_relation(project)
+        with pytest.raises(RuntimeError, match="boom"):
+            adapter.describe_dynamic_table(relation)
+
+        after = _get_qiic_value(adapter)
+        assert after == before