feat(clickhouse-connect): make label mutation optional via database extra

superset/db_engine_specs/base.py

@@ -2112,7 +2112,9 @@ def needs_oauth2(cls, ex: Exception) -> bool:
         return g and hasattr(g, "user") and isinstance(ex, cls.oauth2_exception)
 
     @classmethod
-    def make_label_compatible(cls, label: str) -> str | quoted_name:
+    def make_label_compatible(
+        cls, label: str, database: Optional[Database] = None
+    ) -> str | quoted_name:
         """
         Conditionally mutate and/or quote a sqlalchemy expression label. If
         force_column_alias_quotes is set to True, return the label as a
@@ -2122,9 +2124,15 @@ def make_label_compatible(cls, label: str) -> str | quoted_name:
         generate a truncated label by calling truncate_label().
 
         :param label: expected expression label/alias
+        :param database: optional Database instance for db-specific label mutation logic
         :return: conditionally mutated label supported by the db engine
         """
-        label_mutated = cls._mutate_label(label)
+        if "database" in signature(cls._mutate_label).parameters:
+            # Dynamic dispatch based on subclass signature
+            label_mutated = cls._mutate_label(label, database=database)  # type: ignore[call-arg]
+        else:
+            label_mutated = cls._mutate_label(label)
+
         if (
             cls.max_column_name_length
             and len(label_mutated) > cls.max_column_name_length

superset/db_engine_specs/clickhouse.py

@@ -519,15 +519,41 @@ def validate_parameters(
         return []
 
     @staticmethod
-    def _mutate_label(label: str) -> str:
+    def _mutate_label(label: str, database: Database | None = None) -> str:
         """
-        Suffix with the first six characters from the md5 of the label to avoid
-        collisions with original column names
+        Conditionally suffix the label with the first six characters from the md5
+        of the label to avoid collisions with original column names.
+
+        By default, labels are mutated for backward compatibility. To disable this
+        behavior on a per-database basis, set `mutate_label_name` to `false` in
+        the database's "extra" JSON configuration in the Superset UI:
+
+            {
+                "mutate_label_name": false
+            }
+
+        This is useful when your ClickHouse queries refer to their own aliases
+        within the same SELECT statement.
 
         :param label: Expected expression label
-        :return: Conditionally mutated label
+        :param database: Database instance for db-specific label mutation logic
+        :return: Mutated label if mutation is enabled, otherwise the original label
         """
-        return f"{label}_{hash_from_str(label)[:6]}"
+        mutate_label = True
+
+        if database:
+            try:
+                extra = database.get_extra()
+                mutate_label = extra.get("mutate_label_name", True)
+            except Exception:  # pylint: disable=broad-except
+                logger.warning(
+                    "Error retrieving mutate_label_name setting. Falling back to "
+                    "default behavior of True."
+                )
+
+        if mutate_label:
+            return f"{label}_{hash_from_str(label)[:6]}"
+        return label
 
     @classmethod
     def adjust_engine_params(

superset/models/core.py

@@ -1217,7 +1217,9 @@ def make_sqla_column_compatible(
         label_expected = label or sqla_col.name
         # add quotes to tables
         if self.db_engine_spec.get_allows_alias_in_select(self):
-            label = self.db_engine_spec.make_label_compatible(label_expected)
+            label = self.db_engine_spec.make_label_compatible(
+                label_expected, database=self
+            )
             sqla_col = sqla_col.label(label)
         sqla_col.key = label_expected
         return sqla_col

superset/models/helpers.py

@@ -1013,7 +1013,9 @@ def make_sqla_column_compatible(
         db_engine_spec = self.db_engine_spec
         # add quotes to tables
         if db_engine_spec.get_allows_alias_in_select(self.database):
-            label = db_engine_spec.make_label_compatible(label_expected)
+            label = db_engine_spec.make_label_compatible(
+                label_expected, database=self.database
+            )
             sqla_col = sqla_col.label(label)
         sqla_col.key = label_expected
         return sqla_col
@@ -2684,8 +2686,9 @@ def get_sqla_query(  # pylint: disable=too-many-arguments,too-many-locals,too-ma
         rejected_adhoc_filters_columns: list[Union[str, ColumnTyping]] = []
         applied_adhoc_filters_columns: list[Union[str, ColumnTyping]] = []
         db_engine_spec = self.db_engine_spec
+        # Ensure db-specific label mutation logic can access the current database
         series_column_labels = [
-            db_engine_spec.make_label_compatible(column)
+            db_engine_spec.make_label_compatible(column, database=self.database)
             for column in utils.get_column_names(
                 columns=series_columns or [],
             )
@@ -3301,7 +3304,9 @@ def get_sqla_query(  # pylint: disable=too-many-arguments,too-many-locals,too-ma
                     # in this case the column name, not the alias, needs to be
                     # conditionally mutated, as it refers to the column alias in
                     # the inner query
-                    col_name = db_engine_spec.make_label_compatible(gby_name + "__")
+                    col_name = db_engine_spec.make_label_compatible(
+                        gby_name + "__", database=self.database
+                    )
                     on_clause.append(gby_obj == sa.column(col_name))
 
                 # Use LEFT JOIN when grouping others, INNER JOIN otherwise
@@ -3318,7 +3323,7 @@ def get_sqla_query(  # pylint: disable=too-many-arguments,too-many-locals,too-ma
                     def _create_join_condition(col_name: str, expr: Any) -> Any:
                         # Get the corresponding column from the subquery
                         subq_col_name = db_engine_spec.make_label_compatible(
-                            col_name + "__"
+                            col_name + "__", database=self.database
                         )
                         # Reference the column from the already-created aliased subquery
                         subq_col = subq_alias.c[subq_col_name]

tests/unit_tests/db_engine_specs/test_clickhouse.py

@@ -229,6 +229,33 @@ def test_connect_make_label_compatible(column_name: str, expected_result: str) -
     assert label == expected_result
 
 
+@pytest.mark.parametrize(
+    "extra,column_name,expected_result",
+    [
+        # Default behavior (mutate labels)
+        (None, "time", "time_07cc69"),
+        ({}, "time", "time_07cc69"),
+        # Explicitly enable
+        ({"mutate_label_name": True}, "time", "time_07cc69"),
+        # Disabled (no mutation)
+        ({"mutate_label_name": False}, "time", "time"),
+        ({"mutate_label_name": False}, "count", "count"),
+    ],
+)
+def test_connect_make_label_compatible_with_extra(
+    extra: Optional[dict[str, Any]], column_name: str, expected_result: str
+) -> None:
+    from superset.db_engine_specs.clickhouse import (
+        ClickHouseConnectEngineSpec as spec,  # noqa: N813
+    )
+
+    mock_database = Mock()
+    mock_database.get_extra = Mock(return_value=extra or {})
+
+    label = spec.make_label_compatible(column_name, database=mock_database)
+    assert label == expected_result
+
+
 @pytest.mark.parametrize(
     "schema, expected_result",
     [