[BUGFIX] deduplicate sql metric aliases to prevent view schema collisions (#10926)

great_expectations/execution_engine/sqlalchemy_execution_engine.py

@@ -6,9 +6,7 @@
 import logging
 import math
 import os
-import random
 import re
-import string
 import traceback
 from collections.abc import Generator
 from contextlib import contextmanager
@@ -1176,20 +1174,21 @@ def _organize_metrics_by_domain(  # noqa: C901 # FIXME
                         domain_batches[domain_id] = {"select": [], "metric_ids": []}
                         batch_counters[domain_id] = new_batch_counter
 
-            if self.engine.dialect.name.lower() == GXSqlDialect.CLICKHOUSE:
-                domain_batches[domain_id]["select"].append(
-                    metric_fn.label(
-                        metric_to_resolve.metric_name.join(
-                            random.choices(string.ascii_lowercase, k=4)
-                        )
-                    )
-                )
-                domain_batches[domain_id]["metric_ids"].append(metric_to_resolve.id)
-            else:
-                domain_batches[domain_id]["select"].append(
-                    metric_fn.label(metric_to_resolve.metric_name)
-                )
-                domain_batches[domain_id]["metric_ids"].append(metric_to_resolve.id)
+            alias = metric_to_resolve.metric_name
+
+            # Prevent "Duplicated field name in view schema" SQL errors by deduplicating aliases
+            existing_aliases = {
+                col.name for col in domain_batches[domain_id]["select"] if hasattr(col, "name")
+            }
+
+            if alias in existing_aliases:
+                suffix = 1
+                while f"{alias}_{suffix}" in existing_aliases:
+                    suffix += 1
+                alias = f"{alias}_{suffix}"
+
+            domain_batches[domain_id]["select"].append(metric_fn.label(alias))
+            domain_batches[domain_id]["metric_ids"].append(metric_to_resolve.id)
 
         for domain_id in list(domain_batches.keys()):
             final_domain_id, new_query_entry, new_batch_counter = self._finalize_domain_query(

tests/execution_engine/test_sqlalchemy_metric_alias_deduplication.py

@@ -0,0 +1,64 @@
+from unittest.mock import patch
+
+import pytest
+
+from great_expectations.compatibility.sqlalchemy import sqlalchemy as sa
+from great_expectations.execution_engine.execution_engine import (
+    MetricComputationConfiguration,
+)
+from great_expectations.execution_engine.sqlalchemy_execution_engine import (
+    SqlAlchemyExecutionEngine,
+)
+from great_expectations.validator.metric_configuration import MetricConfiguration
+
+pytestmark = pytest.mark.unit
+
+
+def _make_bundle_entry(column: str) -> MetricComputationConfiguration:
+    """Build a bundle entry whose metric name collides with the others."""
+    return MetricComputationConfiguration(
+        metric_configuration=MetricConfiguration(
+            metric_name="column_values.nonnull.unexpected_count",
+            metric_domain_kwargs={"column": column},
+            metric_value_kwargs=None,
+        ),
+        metric_fn=sa.func.sum(sa.column(column)),
+        compute_domain_kwargs={},
+        accessor_domain_kwargs={"column": column},
+        metric_provider_kwargs={},
+    )
+
+
+def test_organize_metrics_by_domain_deduplicates_metric_aliases():
+    """Regression test for #10926.
+
+    Bundling multiple expectations that resolve to the same underlying metric
+    (e.g. column_values.nonnull.unexpected_count on different columns) must
+    produce unique, deterministic SQL aliases so strict backends (ClickHouse,
+    BigQuery) do not fail with "Duplicated field name in view schema".
+    """
+    engine = SqlAlchemyExecutionEngine(connection_string="sqlite://")
+    selectable = sa.table(
+        "test_table",
+        sa.column("field1"),
+        sa.column("field2"),
+        sa.column("field3"),
+    )
+
+    bundle = [
+        _make_bundle_entry("field1"),
+        _make_bundle_entry("field2"),
+        _make_bundle_entry("field3"),
+    ]
+
+    with patch.object(SqlAlchemyExecutionEngine, "get_domain_records", return_value=selectable):
+        queries = engine._organize_metrics_by_domain(bundle)
+
+    assert len(queries) == 1
+    labels = [select.name for select in queries[0]["select"]]
+    assert labels == [
+        "column_values.nonnull.unexpected_count",
+        "column_values.nonnull.unexpected_count_1",
+        "column_values.nonnull.unexpected_count_2",
+    ]
+    assert len(queries[0]["metric_ids"]) == 3