fix: dataset update uniqueness (apache#25756)

(cherry picked from commit c7f8d11)

superset/daos/dataset.py

@@ -99,11 +99,15 @@ def validate_uniqueness(
 
     @staticmethod
     def validate_update_uniqueness(
-        database_id: int, dataset_id: int, name: str
+        database_id: int,
+        schema: str | None,
+        dataset_id: int,
+        name: str,
     ) -> bool:
         dataset_query = db.session.query(SqlaTable).filter(
             SqlaTable.table_name == name,
             SqlaTable.database_id == database_id,
+            SqlaTable.schema == schema,
             SqlaTable.id != dataset_id,
         )
         return not db.session.query(dataset_query.exists()).scalar()

superset/datasets/commands/update.py

@@ -89,7 +89,10 @@ def validate(self) -> None:
         table_name = self._properties.get("table_name", None)
         # Validate uniqueness
         if not DatasetDAO.validate_update_uniqueness(
-            self._model.database_id, self._model_id, table_name
+            self._model.database_id,
+            self._model.schema,
+            self._model_id,
+            table_name,
         ):
             exceptions.append(DatasetExistsValidationError(table_name))
         # Validate/Populate database not allowed to change

tests/unit_tests/dao/dataset_test.py

@@ -0,0 +1,83 @@
+# 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.
+
+from sqlalchemy.orm.session import Session
+
+from superset.daos.dataset import DatasetDAO
+
+
+def test_validate_update_uniqueness(session: Session) -> None:
+    """
+    Test the `validate_update_uniqueness` static method.
+
+    In particular, allow datasets with the same name in the same database as long as they
+    are in different schemas
+    """
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.core import Database
+
+    SqlaTable.metadata.create_all(session.get_bind())
+
+    database = Database(
+        database_name="my_db",
+        sqlalchemy_uri="sqlite://",
+    )
+    dataset1 = SqlaTable(
+        table_name="my_dataset",
+        schema="main",
+        database=database,
+    )
+    dataset2 = SqlaTable(
+        table_name="my_dataset",
+        schema="dev",
+        database=database,
+    )
+    session.add_all([database, dataset1, dataset2])
+    session.flush()
+
+    # same table name, different schema
+    assert (
+        DatasetDAO.validate_update_uniqueness(
+            database_id=database.id,
+            schema=dataset1.schema,
+            dataset_id=dataset1.id,
+            name=dataset1.table_name,
+        )
+        is True
+    )
+
+    # duplicate schema and table name
+    assert (
+        DatasetDAO.validate_update_uniqueness(
+            database_id=database.id,
+            schema=dataset2.schema,
+            dataset_id=dataset1.id,
+            name=dataset1.table_name,
+        )
+        is False
+    )
+
+    # no schema
+    assert (
+        DatasetDAO.validate_update_uniqueness(
+            database_id=database.id,
+            schema=None,
+            dataset_id=dataset1.id,
+            name=dataset1.table_name,
+        )
+        is True
+    )