fix(sql): make namespace_exists prefix check case-sensitive

pyiceberg/catalog/sql.py

@@ -28,6 +28,7 @@
     String,
     create_engine,
     delete,
+    func,
     insert,
     select,
     text,
@@ -569,15 +570,21 @@ def commit_table(
     def namespace_exists(self, identifier: str | Identifier) -> bool:
         namespace_tuple = Catalog.identifier_to_tuple(identifier)
         namespace = Catalog.namespace_to_string(namespace_tuple, NoSuchNamespaceError)
-        namespace_starts_with = namespace.replace("!", "!!").replace("_", "!_").replace("%", "!%") + ".%"
+        # A namespace exists if it is stored itself or is a parent of a stored
+        # namespace. The prefix comparison must be case-sensitive: LIKE is
+        # case-insensitive for ASCII on SQLite, so it reported e.g. "SALES" as
+        # existing when only "sales.eu" was stored, unlike PostgreSQL. substr
+        # with = is case-sensitive on both backends.
+        namespace_prefix = namespace + "."
+        prefix_length = len(namespace_prefix)
 
         with Session(self.engine) as session:
             stmt = (
                 select(IcebergTables)
                 .where(
                     IcebergTables.catalog_name == self.name,
                     (IcebergTables.table_namespace == namespace)
-                    | (IcebergTables.table_namespace.like(namespace_starts_with, escape="!")),
+                    | (func.substr(IcebergTables.table_namespace, 1, prefix_length) == namespace_prefix),
                 )
                 .limit(1)
             )
@@ -589,7 +596,7 @@ def namespace_exists(self, identifier: str | Identifier) -> bool:
                 .where(
                     IcebergNamespaceProperties.catalog_name == self.name,
                     (IcebergNamespaceProperties.namespace == namespace)
-                    | (IcebergNamespaceProperties.namespace.like(namespace_starts_with, escape="!")),
+                    | (func.substr(IcebergNamespaceProperties.namespace, 1, prefix_length) == namespace_prefix),
                 )
                 .limit(1)
             )

tests/catalog/test_sql.py

@@ -522,6 +522,24 @@ def test_namespace_exists_on_v0_schema(warehouse: Path) -> None:
     assert not catalog.namespace_exists("missing")
 
 
+def test_namespace_exists_is_case_sensitive(warehouse: Path) -> None:
+    """namespace_exists must not match a namespace that differs only in case."""
+    catalog = SqlCatalog(
+        "test",
+        uri=f"sqlite:///{warehouse.as_posix()}/test-ns-exists-case",
+        warehouse=f"file://{warehouse}",
+    )
+    catalog.create_tables()
+    catalog.create_namespace("sales.eu")
+    assert catalog.namespace_exists("sales")
+    assert catalog.namespace_exists("sales.eu")
+    assert not catalog.namespace_exists("SALES")
+    assert not catalog.namespace_exists("SALES.eu")
+    catalog.create_namespace("SALES")
+    assert catalog.namespace_exists("SALES")
+    catalog.destroy_tables()
+
+
 def test_create_and_load_table_on_v0_schema(warehouse: Path) -> None:
     """create_table and load_table should work on V0 schema without iceberg_type column."""
     catalog = _make_v0_catalog(f"sqlite:///{warehouse.as_posix()}/test-v0-create", warehouse)