Fix case-insensitive parent namespace match in SqlCatalog.namespace_exists

pyiceberg/catalog/sql.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Iterable
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -573,31 +574,34 @@ def namespace_exists(self, identifier: str | Identifier) -> bool:
 
         with Session(self.engine) as session:
             stmt = (
-                select(IcebergTables)
+                select(IcebergTables.table_namespace)
                 .where(
                     IcebergTables.catalog_name == self.name,
                     (IcebergTables.table_namespace == namespace)
                     | (IcebergTables.table_namespace.like(namespace_starts_with, escape="!")),
                 )
-                .limit(1)
+                .distinct()
             )
-            result = session.execute(stmt).all()
-            if result:
+            if self._contains_namespace(session.scalars(stmt), namespace):
                 return True
             stmt = (
-                select(IcebergNamespaceProperties)
+                select(IcebergNamespaceProperties.namespace)
                 .where(
                     IcebergNamespaceProperties.catalog_name == self.name,
                     (IcebergNamespaceProperties.namespace == namespace)
                     | (IcebergNamespaceProperties.namespace.like(namespace_starts_with, escape="!")),
                 )
-                .limit(1)
+                .distinct()
             )
-            result = session.execute(stmt).all()
-            if result:
+            if self._contains_namespace(session.scalars(stmt), namespace):
                 return True
         return False
 
+    @staticmethod
+    def _contains_namespace(candidates: Iterable[str], namespace: str) -> bool:
+        # LIKE is case-insensitive on some backends (e.g. SQLite), so re-check the prefix match
+        return any(candidate == namespace or candidate.startswith(f"{namespace}.") for candidate in candidates)
+
     @override
     def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
         """Create a namespace in the catalog.

tests/catalog/test_catalog_behaviors.py

@@ -1046,6 +1046,9 @@ def test_namespace_exists(catalog: Catalog) -> None:
     assert not catalog.namespace_exists("db_")
     # make sure '%' is escaped in the query
     assert not catalog.namespace_exists("db%")
+    # namespaces are case-sensitive, also when matching a parent namespace
+    assert not catalog.namespace_exists("DB2")
+    assert not catalog.namespace_exists("db3.NS1")
 
 
 # Namespace properties