From 9e15ea0baad14695ef8222fe10bb70274944d75e Mon Sep 17 00:00:00 2001
From: Dhruv Gupta <todhruvg@gmail.com>
Date: Tue, 27 Oct 2026 06:30:38 -0400
Subject: [PATCH] Fix case-insensitive parent namespace match in
 SqlCatalog.namespace_exists

The parent-namespace check uses LIKE, which SQLite evaluates
case-insensitively for ASCII. With only `sales.eu` present,
`namespace_exists("SALES")` returned True, so `create_table("SALES.t")`
succeeded in a namespace that does not exist and `create_namespace("SALES")`
failed with NamespaceAlreadyExistsError. Re-check the LIKE matches with a
case-sensitive prefix comparison so the result matches the exact-equality
lookups used elsewhere in the catalog.
---
 pyiceberg/catalog/sql.py                | 20 ++++++++++++--------
 tests/catalog/test_catalog_behaviors.py |  3 +++
 2 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py
index 94c7965a88..9526771a7b 100644
--- a/pyiceberg/catalog/sql.py
+++ b/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.
diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py
index 1148ca8527..5a7bf67bb0 100644
--- a/tests/catalog/test_catalog_behaviors.py
+++ b/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