fix(datasets): return 400 not 500 when get_or_create matches ambiguous catalog rows

The `if schema:` branch of `DatasetRestApi.get_or_create_dataset` called
`DatasetDAO.get_table_by_catalog_schema_and_name` (ending in `.one_or_none()`)
without guarding `MultipleResultsFound`. The sibling no-schema branch already
catches this and returns a clean 400 (added in cb1694575c for #30377), but the
schema branch from that same commit was left unguarded.

This is reachable: the composite unique constraint
`(database_id, catalog, schema, table_name)` treats NULL catalogs as distinct
in Postgres/MySQL, so two legacy datasets sharing (database_id, schema,
table_name) with catalog=None do not violate it. Calling get_or_create with an
explicit schema then 500s with a raw MultipleResultsFound traceback.

Mirror the existing no-schema guard for the schema branch, returning a 400 with
an actionable message pointing at the catalog field. Adds a regression test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset/datasets/api.py

@@ -1536,7 +1536,10 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
         # physical Postgres/MySQL table stored with a non-NULL schema).
         # If two datasets share the ``table_name`` across schemas and the
         # caller omits ``schema``, surface a 400 with an actionable message
-        # instead of the original 500 ``MultipleResultsFound``.
+        # instead of the original 500 ``MultipleResultsFound``. The same guard
+        # applies when the caller supplies ``schema`` but two legacy rows still
+        # match with ``catalog=None`` (the composite unique constraint treats
+        # NULL catalogs as distinct), so both branches catch the exception.
         # Catalog follows the same literal-pass rule: existing datasets
         # created before multi-catalog support landed are stored with
         # ``catalog=None``, so applying ``database.get_default_catalog()``
@@ -1544,9 +1547,20 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
         schema = body.get("schema") or None
         catalog = body.get("catalog") or None
         if schema:
-            table = DatasetDAO.get_table_by_catalog_schema_and_name(
-                database_id, schema, table_name, catalog=catalog
-            )
+            try:
+                table = DatasetDAO.get_table_by_catalog_schema_and_name(
+                    database_id, schema, table_name, catalog=catalog
+                )
+            except MultipleResultsFound:
+                return self.response_400(
+                    message=(
+                        f"Multiple datasets named '{table_name}' exist in "
+                        f"schema '{schema}' of this database, differing only "
+                        "by catalog. Specify the 'catalog' field to "
+                        "disambiguate, or contact an admin about removing "
+                        "duplicate legacy datasets."
+                    )
+                )
         else:
             try:
                 table = DatasetDAO.get_table_by_name(database_id, table_name)

tests/integration_tests/datasets/api_tests.py

@@ -3269,6 +3269,61 @@ class TestDatasetApi(SupersetTestCase):
         assert rv.status_code == 400
         assert "Specify the 'schema' field" in rv.data.decode("utf-8")
 
+    def test_get_or_create_dataset_with_schema_returns_400_when_ambiguous(self):
+        """
+        Dataset API: regression for the ``if schema:`` branch of
+        ``get_or_create``.
+
+        Two legacy datasets can share ``(database_id, schema, table_name)``
+        while both carry ``catalog=None``: the composite unique constraint
+        treats NULL catalogs as distinct, so it does not prevent this. When
+        the caller supplies ``schema`` and two such rows match,
+        ``one_or_none()`` raises ``MultipleResultsFound``. The API must return
+        a 400 with an actionable message rather than 500-ing — mirroring the
+        no-schema guard.
+        """
+        if get_main_database().backend == "sqlite":
+            pytest.skip(
+                "SQLite has a legacy single-column unique constraint on "
+                "table_name that prevents seeding two same-name datasets in "
+                "the same schema"
+            )
+
+        self.login(ADMIN_USERNAME)
+        admin_id = self.get_user("admin").id
+        examples_db = get_example_database()
+        table_name = "test_get_or_create_ambiguous_catalog"
+        schema = "same_schema"
+
+        # Both rows share database_id + schema + table_name with catalog=None
+        # (the default), so they match the ``if schema:`` lookup ambiguously.
+        ds_a = self.insert_dataset(
+            table_name,
+            [admin_id],
+            examples_db,
+            schema=schema,
+            fetch_metadata=False,
+        )
+        ds_b = self.insert_dataset(
+            table_name,
+            [admin_id],
+            examples_db,
+            schema=schema,
+            fetch_metadata=False,
+        )
+        self.items_to_delete = [ds_a, ds_b]
+
+        rv = self.client.post(
+            "api/v1/dataset/get_or_create/",
+            json={
+                "table_name": table_name,
+                "schema": schema,
+                "database_id": examples_db.id,
+            },
+        )
+        assert rv.status_code == 400
+        assert "catalog" in rv.data.decode("utf-8")
+
     @pytest.mark.usefixtures(
         "load_energy_table_with_slice", "load_birth_names_dashboard_with_slices"
     )