fix SqlCatalog list_namespaces()

Author: alessandro-nori

pyiceberg/catalog/sql.py

@@ -610,15 +610,29 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         table_stmt = select(IcebergTables.table_namespace).where(IcebergTables.catalog_name == self.name)
         namespace_stmt = select(IcebergNamespaceProperties.namespace).where(IcebergNamespaceProperties.catalog_name == self.name)
         if namespace:
-            namespace_str = Catalog.namespace_to_string(namespace, NoSuchNamespaceError)
-            table_stmt = table_stmt.where(IcebergTables.table_namespace.like(namespace_str))
-            namespace_stmt = namespace_stmt.where(IcebergNamespaceProperties.namespace.like(namespace_str))
+            namespace_like = Catalog.namespace_to_string(namespace, NoSuchNamespaceError) + "%"
+            table_stmt = table_stmt.where(IcebergTables.table_namespace.like(namespace_like))
+            namespace_stmt = namespace_stmt.where(IcebergNamespaceProperties.namespace.like(namespace_like))
         stmt = union(
             table_stmt,
             namespace_stmt,
         )
         with Session(self.engine) as session:
-            return [Catalog.identifier_to_tuple(namespace_col) for namespace_col in session.execute(stmt).scalars()]
+            namespaces = [Catalog.identifier_to_tuple(namespace_col) for namespace_col in session.execute(stmt).scalars()]
+
+            sub_namespaces_level_length = 1
+            if namespace:
+                namespace_tuple = Catalog.identifier_to_tuple(namespace)
+                sub_namespaces_level_length = len(namespace_tuple) + 1
+
+            # only get sub namespaces/children
+            result = list({ns[:sub_namespaces_level_length] for ns in namespaces if len(ns) >= sub_namespaces_level_length})
+
+            if namespace:
+                # exclude fuzzy matches when `namespace` contains `%` or `_`
+                result = [ns for ns in result if ns[: len(namespace_tuple)] == namespace_tuple]
+
+            return result
 
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """Get properties for a namespace.

tests/catalog/test_sql.py

@@ -17,7 +17,7 @@
 
 import os
 from pathlib import Path
-from typing import Any, Generator, List, cast
+from typing import Any, Generator, cast
 
 import pyarrow as pa
 import pytest
@@ -1116,17 +1116,31 @@ def test_create_namespace_with_empty_identifier(catalog: SqlCatalog, empty_names
         lazy_fixture("catalog_sqlite"),
     ],
 )
-@pytest.mark.parametrize("namespace_list", [lazy_fixture("database_list"), lazy_fixture("hierarchical_namespace_list")])
-def test_list_namespaces(catalog: SqlCatalog, namespace_list: List[str]) -> None:
+def test_list_namespaces(catalog: SqlCatalog) -> None:
+    namespace_list = ["db", "db.ns1", "db.ns1.ns2", "db.ns2", "db2", "db2.ns1", "db%"]
     for namespace in namespace_list:
         catalog.create_namespace(namespace)
-    # Test global list
+
     ns_list = catalog.list_namespaces()
-    for namespace in namespace_list:
-        assert Catalog.identifier_to_tuple(namespace) in ns_list
-        # Test individual namespace list
-        assert len(one_namespace := catalog.list_namespaces(namespace)) == 1
-        assert Catalog.identifier_to_tuple(namespace) == one_namespace[0]
+    expected_list = [("db",), ("db2",), ("db%",)]
+    assert len(ns_list) == len(expected_list)
+    for ns in expected_list:
+        assert ns in ns_list
+
+    ns_list = catalog.list_namespaces("db")
+    expected_list = [("db", "ns1"), ("db", "ns2")]
+    assert len(ns_list) == len(expected_list)
+    for ns in expected_list:
+        assert ns in ns_list
+
+    ns_list = catalog.list_namespaces("db.ns1")
+    expected_list = [("db", "ns1", "ns2")]
+    assert len(ns_list) == len(expected_list)
+    for ns in expected_list:
+        assert ns in ns_list
+
+    ns_list = catalog.list_namespaces("db.ns1.ns2")
+    assert len(ns_list) == 0
 
 
 @pytest.mark.parametrize(