stop setting quoted_identifiers_ignore_case (#1159)

Author: colin-rogers-dbt

dbt-snowflake/.changes/unreleased/Under the Hood-20250616-155209.yaml

@@ -0,0 +1,7 @@
+kind: Under the Hood
+body: remove reliance on setting session `quoted_identifiers_ignore_case` parameter
+  to normalize snowflake metadata queries
+time: 2025-06-16T15:52:09.002616-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1159"

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

@@ -276,7 +276,9 @@ def list_relations_without_caching(
             raise
 
         columns = ["database_name", "schema_name", "name", "kind", "is_dynamic", "is_iceberg"]
-
+        schema_objects = schema_objects.rename(
+            column_names=[col.lower() for col in schema_objects.column_names]
+        )
         return [self._parse_list_relations_result(obj) for obj in schema_objects.select(columns)]
 
     def _parse_list_relations_result(self, result: "agate.Row") -> SnowflakeRelation:
@@ -482,58 +484,34 @@ def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelat
         return None
 
     @available
-    def describe_dynamic_table(self, relation: RelationConfig) -> Dict[str, Any]:
+    def describe_dynamic_table(self, relation: SnowflakeRelation) -> 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}"
+        quoting = relation.quote_policy
+        schema = f'"{relation.schema}"' if quoting.schema else relation.schema
+        database = f'"{relation.database}"' if quoting.database else relation.database
+        show_sql = (
+            f"show dynamic tables like '{relation.identifier}' in schema {database}.{schema}"
+        )
+        res, dt_table = self.execute(show_sql, fetch=True)
+        if res.code != "SUCCESS":
+            raise DbtRuntimeError(f"Could not get dynamic query metadata: {show_sql} failed")
+        # normalize column names to lower case, this still preserves column order
+        dt_table = dt_table.rename(column_names=[name.lower() for name in dt_table.column_names])
+        return {
+            "dynamic_table": dt_table.select(
+                [
+                    "name",
+                    "schema_name",
+                    "database_name",
+                    "text",
+                    "target_lag",
+                    "warehouse",
+                    "refresh_mode",
+                ]
             )
-            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/metadata/list_relations_without_caching.sql

@@ -17,8 +17,6 @@
 
     {%- set paginated_state = namespace(paginated_results=[], watermark=none) -%}
 
-    {%- do run_query('alter session set quoted_identifiers_ignore_case = false;') -%}
-
     {#-
         loop an extra time to catch the breach of max iterations
         Note: while range is 0-based, loop.index starts at 1
@@ -46,8 +44,6 @@
 
     {%- endfor -%}
 
-    {%- do run_query('alter session unset quoted_identifiers_ignore_case;') -%}
-
     {#- grab the first table in the paginated results to access the `merge` method -#}
     {%- set agate_table = paginated_state.paginated_results[0] -%}
     {%- do return(agate_table.merge(paginated_state.paginated_results)) -%}