fix(snowflake): quote case-sensitive identifiers in adhoc SIMPLE metrics

A SIMPLE adhoc metric could still emit an unquoted physical column, so
Snowflake upper-cased the identifier and failed to resolve exact-case
columns. Two paths bypassed `prepare_identifier`:

- `SqlaTable.adhoc_metric_to_sqla` only routed through `get_sqla_col`
  when the column was present in `columns_by_name`; the fallback built a
  bare `column()`.
- `ExploreMixin.adhoc_metric_to_sqla` built a bare `sa.column()`
  unconditionally, so every SIMPLE adhoc metric on the query-object path
  was affected, not just a fallback case.

Both now prepare the identifier the same way as the existing call sites.
The SQL expression branches are unchanged; they already route through
`_process_select_expression`/`literal_column`.

superset/connectors/sqla/models.py

@@ -1235,7 +1235,11 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
-            col = column(self.column_name, type_=type_)
+            identifier = db_engine_spec.prepare_identifier(
+                cast(str, self.column_name),
+                normalize_columns=bool(getattr(self.table, "normalize_columns", False)),
+            )
+            col = column(identifier, type_=type_)
         col = self.database.make_sqla_column_compatible(col, label)
         return col
 
@@ -1265,12 +1269,15 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
 
         pdf = self.python_date_format
         is_epoch = pdf in ("epoch_s", "epoch_ms")
-        column_spec = self.db_engine_spec.get_column_spec(
-            self.type, db_extra=self.db_extra
-        )
+        db_engine_spec = self.db_engine_spec
+        column_spec = db_engine_spec.get_column_spec(self.type, db_extra=self.db_extra)
         type_ = column_spec.sqla_type if column_spec else DateTime
         if not self.expression and not time_grain and not is_epoch:
-            sqla_col = column(self.column_name, type_=type_)
+            identifier = db_engine_spec.prepare_identifier(
+                cast(str, self.column_name),
+                normalize_columns=bool(getattr(self.table, "normalize_columns", False)),
+            )
+            sqla_col = column(identifier, type_=type_)
             return self.database.make_sqla_column_compatible(sqla_col, label)
         if expression := self.expression:
             if template_processor:
@@ -1295,7 +1302,11 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
-            col = column(self.column_name, type_=type_)
+            identifier = db_engine_spec.prepare_identifier(
+                cast(str, self.column_name),
+                normalize_columns=bool(getattr(self.table, "normalize_columns", False)),
+            )
+            col = column(identifier, type_=type_)
         if (
             apply_dataset_offset
             and time_grain
@@ -1932,7 +1943,12 @@ class SqlaTable(
                     template_processor=template_processor
                 )
             else:
-                sqla_column = column(column_name)
+                sqla_column = column(
+                    self.db_engine_spec.prepare_identifier(
+                        column_name,
+                        normalize_columns=bool(self.normalize_columns),
+                    )
+                )
 
             if isinstance(aggregate, str) and aggregate in self.sqla_aggregations:
                 sqla_metric = self.sqla_aggregations[aggregate](sqla_column)

superset/db_engine_specs/base.py

@@ -2959,6 +2959,19 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
 
         return name
 
+    @classmethod
+    def prepare_identifier(
+        cls,
+        name: str,
+        normalize_columns: bool = False,
+    ) -> str:
+        """
+        Prepare a physical identifier for SQLAlchemy column construction.
+
+        The default preserves SQLAlchemy's automatic identifier-quoting behavior.
+        """
+        return name
+
     @classmethod
     def quote_table(cls, table: Table, dialect: Dialect) -> str:
         """

superset/db_engine_specs/snowflake.py

@@ -34,6 +34,7 @@ from sqlalchemy import text, types
 from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import URL
 from sqlalchemy.exc import DatabaseError as SqlalchemyDatabaseError
+from sqlalchemy.sql import quoted_name
 from sqlalchemy.sql.elements import ColumnElement
 
 from superset import is_feature_enabled, security_manager
@@ -156,6 +157,17 @@ class SnowflakeEngineSpec(PostgresBaseEngineSpec):
     supports_catalog = supports_dynamic_catalog = supports_cross_catalog_queries = True
     supports_grouping_sets = True
 
+    @classmethod
+    def prepare_identifier(
+        cls,
+        name: str,
+        normalize_columns: bool = False,
+    ) -> str:
+        """Preserve exact-case physical identifiers when columns are not normalized."""
+        if normalize_columns:
+            return name
+        return quoted_name(name, quote=True)
+
     metadata = {
         "description": "Snowflake is a cloud-native data warehouse.",
         "logo": "snowflake.svg",

superset/models/helpers.py

@@ -3563,7 +3563,12 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             aggregate: Any = metric.get("aggregate")
             metric_column = metric.get("column") or {}
             column_name = cast(str, metric_column.get("column_name"))
-            sqla_column = sa.column(column_name)
+            sqla_column = sa.column(
+                self.db_engine_spec.prepare_identifier(
+                    column_name,
+                    normalize_columns=bool(self.normalize_columns),
+                )
+            )
 
             if isinstance(aggregate, str) and aggregate in self.sqla_aggregations:
                 sqla_metric = self.sqla_aggregations[aggregate](sqla_column)
@@ -4269,7 +4274,11 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
-            col = sa.column(tbl_column.column_name, type_=type_)
+            identifier = db_engine_spec.prepare_identifier(
+                cast(str, tbl_column.column_name),
+                normalize_columns=bool(self.normalize_columns),
+            )
+            col = sa.column(identifier, type_=type_)
         col = self.make_sqla_column_compatible(col, label)
         return col
 

tests/unit_tests/connectors/sqla/models_test.py

@@ -21,6 +21,7 @@ import pandas as pd
 import pytest
 from pytest_mock import MockerFixture
 from sqlalchemy import create_engine
+from sqlalchemy.dialects import sqlite
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.session import Session
 
@@ -45,10 +46,121 @@ from superset.models.helpers import (
     validate_rendered_expression,
 )
 from superset.sql.parse import Table
-from superset.superset_typing import QueryObjectDict
+from superset.superset_typing import AdhocMetric, QueryObjectDict
 from superset.utils import json
 
 
+def test_get_sqla_col_quotes_snowflake_case_sensitive_identifier(
+    mocker: MockerFixture,
+) -> None:
+    """Snowflake physical columns retain their exact reflected case in generated SQL."""
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+    database = Database(database_name="db", sqlalchemy_uri="sqlite://")
+    mocker.patch.object(
+        Database,
+        "get_db_engine_spec",
+        return_value=SnowflakeEngineSpec,
+    )
+    table = SqlaTable(
+        table_name="bug_test",
+        database=database,
+        normalize_columns=False,
+    )
+    tbl_column = TableColumn(column_name="id", type="INTEGER", table=table)
+
+    rendered = str(
+        tbl_column.get_sqla_col().compile(
+            dialect=sqlite.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert rendered == '"id"'
+
+
+@pytest.mark.parametrize("time_grain", [None, "P1D"])
+def test_get_timestamp_expression_quotes_snowflake_case_sensitive_identifier(
+    mocker: MockerFixture,
+    time_grain: str | None,
+) -> None:
+    """Snowflake timestamp paths quote exact-case physical columns."""
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+    database = Database(database_name="db", sqlalchemy_uri="sqlite://")
+    mocker.patch.object(
+        Database,
+        "get_db_engine_spec",
+        return_value=SnowflakeEngineSpec,
+    )
+    table = SqlaTable(
+        table_name="bug_test",
+        database=database,
+        normalize_columns=False,
+    )
+    tbl_column = TableColumn(
+        column_name="created_at",
+        type="TIMESTAMP",
+        table=table,
+    )
+
+    rendered = str(
+        tbl_column.get_timestamp_expression(time_grain=time_grain).compile(
+            dialect=sqlite.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert '"created_at"' in rendered
+
+
+def test_adhoc_metric_to_sqla_quotes_snowflake_column_absent_from_columns_by_name(
+    mocker: MockerFixture,
+) -> None:
+    """A SIMPLE adhoc metric quotes exact-case Snowflake columns even when the
+    metric's column is unknown to the dataset.
+
+    ``adhoc_metric_to_sqla`` only routes through ``TableColumn.get_sqla_col`` when
+    the column is present in ``columns_by_name``; the fallback builds a bare
+    ``column()`` and must apply the same identifier preparation, otherwise
+    SQLAlchemy upper-cases the unquoted name and Snowflake fails to resolve it.
+    """
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+    database = Database(database_name="db", sqlalchemy_uri="sqlite://")
+    mocker.patch.object(
+        Database,
+        "get_db_engine_spec",
+        return_value=SnowflakeEngineSpec,
+    )
+    table = SqlaTable(
+        table_name="bug_test",
+        database=database,
+        normalize_columns=False,
+    )
+    metric: AdhocMetric = {
+        "expressionType": "SIMPLE",
+        "aggregate": "SUM",
+        "column": {"column_name": "amount"},
+        "label": "total",
+    }
+
+    # Deliberately empty so the lookup misses and the fallback branch runs.
+    sqla_metric = table.adhoc_metric_to_sqla(metric, {})
+
+    rendered = str(
+        sqla_metric.compile(
+            dialect=sqlite.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert '"amount"' in rendered, (
+        f"Expected the exact-case column to be quoted, got: {rendered}"
+    )
+    assert "(amount)" not in rendered, f"Column was aggregated unquoted: {rendered}"
+
+
 def test_query_bubbles_errors(mocker: MockerFixture) -> None:
     """
     Test that the `query` method bubbles exceptions correctly.

tests/unit_tests/db_engine_specs/test_base.py

@@ -291,6 +291,13 @@ def test_get_default_catalog(mocker: MockerFixture) -> None:
     assert BaseEngineSpec.get_default_catalog(database) is None
 
 
+def test_prepare_identifier_returns_name_unchanged() -> None:
+    name = "physical_column"
+
+    assert BaseEngineSpec.prepare_identifier(name, normalize_columns=False) is name
+    assert BaseEngineSpec.prepare_identifier(name, normalize_columns=True) is name
+
+
 def test_quote_table() -> None:
     """
     Test the `quote_table` function.

tests/unit_tests/db_engine_specs/test_snowflake.py

@@ -24,6 +24,7 @@ from unittest import mock
 import pytest
 from pytest_mock import MockerFixture
 from sqlalchemy.engine.url import make_url, URL
+from sqlalchemy.sql import quoted_name
 
 from superset.app import SupersetApp
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
@@ -33,6 +34,32 @@ from tests.unit_tests.db_engine_specs.utils import assert_convert_dttm
 from tests.unit_tests.fixtures.common import dttm  # noqa: F401
 
 
+@pytest.mark.parametrize("name", ["lowercase", "UPPERCASE"])
+def test_prepare_identifier_quotes_exact_case_names(name: str) -> None:
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+    identifier = SnowflakeEngineSpec.prepare_identifier(
+        name,
+        normalize_columns=False,
+    )
+
+    assert isinstance(identifier, quoted_name)
+    assert str(identifier) == name
+    assert identifier.quote is True
+
+
+def test_prepare_identifier_preserves_normalized_name() -> None:
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+    name = "lowercase"
+    identifier = SnowflakeEngineSpec.prepare_identifier(
+        name,
+        normalize_columns=True,
+    )
+
+    assert identifier is name
+
+
 @pytest.mark.parametrize(
     "target_type,expected_result",
     [

tests/unit_tests/models/helpers_test.py

@@ -4619,6 +4619,83 @@ def test_simple_metric_quotes_column_requiring_quoting(database: Database) -> No
     )
 
 
+def test_explore_mixin_adhoc_metric_quotes_snowflake_case_sensitive_identifier(
+    database: Database,
+) -> None:
+    """``ExploreMixin.adhoc_metric_to_sqla`` quotes exact-case Snowflake columns.
+
+    Unlike the ``SqlaTable`` override, this implementation builds the aggregate
+    from a bare ``sa.column()`` unconditionally, so every SIMPLE adhoc metric on
+    the query-object path bypassed identifier preparation.
+    """
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    datasource = MagicMock()
+    datasource.database = database
+    datasource.db_engine_spec = SnowflakeEngineSpec
+    datasource.normalize_columns = False
+    datasource.sqla_aggregations = ExploreMixin.sqla_aggregations
+    for method in ("adhoc_metric_to_sqla", "make_sqla_column_compatible"):
+        setattr(datasource, method, getattr(ExploreMixin, method).__get__(datasource))
+
+    metric: AdhocMetric = {
+        "expressionType": "SIMPLE",
+        "aggregate": "SUM",
+        "column": {"column_name": "amount"},
+        "label": "total",
+    }
+
+    with database.get_sqla_engine() as engine:
+        dialect = engine.dialect
+
+    rendered = str(
+        datasource.adhoc_metric_to_sqla(metric, {}).compile(
+            dialect=dialect,
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert '"amount"' in rendered, (
+        f"Expected the exact-case column to be quoted, got: {rendered}"
+    )
+    assert "(amount)" not in rendered, f"Column was aggregated unquoted: {rendered}"
+
+
+def test_convert_tbl_column_quotes_snowflake_case_sensitive_identifier(
+    database: Database,
+    mocker: MockerFixture,
+) -> None:
+    """The chart query-object path quotes exact-case Snowflake physical columns."""
+    from superset.connectors.sqla.models import SqlaTable, TableColumn
+    from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+    from superset.models.core import Database
+
+    mocker.patch.object(
+        Database,
+        "get_db_engine_spec",
+        return_value=SnowflakeEngineSpec,
+    )
+    table = SqlaTable(
+        database=database,
+        table_name="bug_test",
+        normalize_columns=False,
+    )
+    tbl_column = TableColumn(column_name="name", type="VARCHAR", table=table)
+
+    with database.get_sqla_engine() as engine:
+        dialect = engine.dialect
+
+    rendered = str(
+        table.convert_tbl_column_to_sqla_col(tbl_column).compile(
+            dialect=dialect,
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert rendered == '"name"'
+
+
 @pytest.mark.parametrize(
     "native_type",
     [