fix(migrations): actually drop _customer_location_uc (list == set no-op)

superset/migrations/versions/2026-07-31_10-00_16755d4ca4ae_drop__customer_location_uc_take_2.py

@@ -0,0 +1,91 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Drop _customer_location_uc (take 2)
+
+Revision ID: 16755d4ca4ae
+Revises: e7d93a524ff6
+Create Date: 2026-07-31 10:00:00.000000
+"""
+
+import logging
+
+from alembic import op
+from migration_utils import create_unique_constraint, drop_unique_constraint
+from sqlalchemy.engine.reflection import Inspector
+
+from superset.utils.core import generic_find_uq_constraint_name
+
+# revision identifiers, used by Alembic.
+revision = "16755d4ca4ae"
+down_revision = "e7d93a524ff6"
+
+logger = logging.getLogger("alembic.env")
+
+
+def upgrade():
+    """Re-attempt the constraint drop that migration ``df3d7e2eb9a4``
+    intended but silently skipped.
+
+    That migration passed a **list** to ``generic_find_uq_constraint_name``,
+    whose body compared ``columns == set(uq["column_names"])`` — and
+    ``list == set`` is always ``False`` in Python, so the constraint was
+    never found and never dropped. Databases migrated through it still
+    carry the legacy 3-column constraint ``(database_id, schema,
+    table_name)``, which both leaks across catalogs (no ``catalog`` leg)
+    and diverges from schemas built from model metadata, where the model's
+    4-column constraint exists instead.
+
+    The lookup below passes a set (and the helper itself coerces, so the
+    foot-gun is gone either way). On databases where the legacy constraint
+    is already absent the lookup finds nothing and the migration is a
+    harmless no-op. The model's intended 4-column constraint can never
+    match a 3-column set comparison, so it is not at risk.
+    """
+    bind = op.get_bind()
+    inspector = Inspector.from_engine(bind)
+
+    if constraint_name := generic_find_uq_constraint_name(
+        "tables",
+        {"database_id", "schema", "table_name"},
+        inspector,
+    ):
+        drop_unique_constraint(op, constraint_name, "tables")
+
+
+def downgrade():
+    """Restore the legacy constraint, mirroring ``df3d7e2eb9a4``'s
+    downgrade. Best-effort: rows written after the drop may legitimately
+    collide on ``(database_id, schema, table_name)`` across catalogs, in
+    which case the constraint cannot be recreated and the downgrade
+    proceeds with a warning — matching the tolerant posture of the
+    constraint's original creator, which wrapped creation in try/except.
+    """
+    try:
+        create_unique_constraint(
+            op,
+            "_customer_location_uc",
+            "tables",
+            ["database_id", "schema", "table_name"],
+        )
+    except Exception:  # pylint: disable=broad-except
+        logger.warning(
+            "Could not recreate _customer_location_uc on 'tables'; existing "
+            "rows likely collide on (database_id, schema, table_name) across "
+            "catalogs. Continuing without the legacy constraint.",
+            exc_info=True,
+        )

superset/utils/core.py

@@ -36,7 +36,7 @@
 import uuid
 import warnings
 import zlib
-from collections.abc import Iterable, Iterator, Sequence
+from collections.abc import Collection, Iterable, Iterator, Sequence
 from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from datetime import timedelta
@@ -702,12 +702,19 @@ def generic_find_fk_constraint_names(  # pylint: disable=invalid-name
 
 
 def generic_find_uq_constraint_name(
-    table: str, columns: set[str], insp: Inspector
+    table: str, columns: Collection[str], insp: Inspector
 ) -> str | None:
-    """Utility to find a unique constraint name in alembic migrations"""
+    """Utility to find a unique constraint name in alembic migrations.
 
+    ``columns`` is coerced to a set before comparison. Historically the
+    parameter was annotated ``set[str]`` but compared with ``==`` against a
+    set — a caller passing a list (as migration ``df3d7e2eb9a4`` did)
+    silently never matched, because ``list == set`` is always ``False``.
+    Coercing removes that foot-gun for future callers.
+    """
+    target = set(columns)
     for uq in insp.get_unique_constraints(table):
-        if columns == set(uq["column_names"]):
+        if target == set(uq["column_names"]):
             return uq["name"]
 
     return None

tests/unit_tests/utils/test_core.py

@@ -36,6 +36,7 @@
     FilterOperator,
     generic_find_constraint_name,
     generic_find_fk_constraint_name,
+    generic_find_uq_constraint_name,
     get_datasource_full_name,
     get_query_source_from_request,
     get_stacktrace,
@@ -700,6 +701,71 @@ def test_generic_find_fk_constraint_none_exist():
     assert result is None
 
 
+def test_generic_find_uq_constraint_accepts_list():
+    """Regression pin for the ``list == set`` foot-gun (sc-112173).
+
+    Migration ``df3d7e2eb9a4`` passed a list and silently never matched,
+    because the helper compared it with ``==`` against a set. The helper
+    coerces its ``columns`` argument, so a list argument MUST find the
+    constraint."""
+    insp_mock = MagicMock()
+    insp_mock.get_unique_constraints.return_value = [
+        {
+            "name": "_customer_location_uc",
+            "column_names": ["database_id", "schema", "table_name"],
+        },
+    ]
+
+    result = generic_find_uq_constraint_name(
+        "tables",
+        ["database_id", "schema", "table_name"],  # deliberately a list
+        insp_mock,
+    )
+
+    assert result == "_customer_location_uc"
+
+
+def test_generic_find_uq_constraint_with_set():
+    """The documented set-shaped argument keeps working unchanged."""
+    insp_mock = MagicMock()
+    insp_mock.get_unique_constraints.return_value = [
+        {
+            "name": "_customer_location_uc",
+            "column_names": ["database_id", "schema", "table_name"],
+        },
+    ]
+
+    result = generic_find_uq_constraint_name(
+        "tables",
+        {"database_id", "schema", "table_name"},
+        insp_mock,
+    )
+
+    assert result == "_customer_location_uc"
+
+
+def test_generic_find_uq_constraint_no_partial_match():
+    """A 3-column lookup MUST NOT match a 4-column constraint: the
+    take-2 drop migration relies on exact set equality so the model's
+    intended ``(database_id, catalog, schema, table_name)`` constraint is
+    never at risk."""
+    insp_mock = MagicMock()
+    insp_mock.get_unique_constraints.return_value = [
+        {
+            "name": "uq_tables_database_id",
+            "column_names": ["database_id", "catalog", "schema", "table_name"],
+        },
+    ]
+
+    result = generic_find_uq_constraint_name(
+        "tables",
+        {"database_id", "schema", "table_name"},
+        insp_mock,
+    )
+
+    assert result is None
+
+
 def test_get_datasource_full_name():
     """
     Test the `get_datasource_full_name` function.