fix(iceberg): listnamespaces includes warehouse name as root (#12761)

Author: chakru-r

metadata-service/iceberg-catalog/pyproject.toml

@@ -44,3 +44,9 @@ disallow_untyped_defs = false
 
 [tool.pyright]
 extraPaths = ['tests']
+
+[tool.pytest.ini_options]
+markers = [
+	"quick: marks limited tests for a quick validation",
+	"serial",
+]

metadata-service/iceberg-catalog/src/integrationTest/integration_test.py

@@ -212,6 +212,9 @@ def test_iceberg_quick(spark_session, warehouse):
     _test_basic_view_ops(spark_session)
     _test_rename_ops(spark_session)
 
+    result = spark_session.sql("show namespaces")
+    assert (result[result["namespace"] == "default"].count() == 1)
+
 
 def _create_table(spark_session, ns, table_name):
     spark_session.sql("create namespace if not exists default")
@@ -221,12 +224,19 @@ def _create_table(spark_session, ns, table_name):
 
     spark_session.sql(f"insert into {ns}.{table_name} values (1, 'foo' ) ")
 
+    result = spark_session.sql("show namespaces")
+    assert (result[result["namespace"] == "default"].count() == 1)
+
+    result = spark_session.sql("show namespaces in default")
+    assert (result[result["namespace"] == f"{ns}"].count() == 1)
+
+
 
 def test_load_tables(spark_session, warehouse):
-    namespace_count = 3
-    table_count = 4
+    namespace_count = 2
+    table_count = 2
     for ns_index in range(namespace_count):
-        ns = f"default_ns{ns_index}"
+        ns = f"default.ns{ns_index}"
         for table_index in range(table_count):
             table_name = f"table_{table_index}"
             _create_table(spark_session, ns, table_name)

metadata-service/iceberg-catalog/src/main/java/io/datahubproject/iceberg/catalog/Utils.java

@@ -75,6 +75,10 @@ public static String parentDir(String fileLocation) {
   }
 
   public static String namespaceNameFromContainerUrn(Urn urn) {
-    return urn.toString().substring(NAMESPACE_CONTAINER_PREFIX.length());
+    // Must do inverse of implementation of method containerUrn(String platformInstance, String[]
+    // levels) in this file
+    String namespaceWithPlatformInstance =
+        urn.toString().substring(NAMESPACE_CONTAINER_PREFIX.length());
+    return namespaceWithPlatformInstance.substring(namespaceWithPlatformInstance.indexOf('.') + 1);
   }
 }

metadata-service/iceberg-catalog/src/test/java/io/datahubproject/iceberg/catalog/DataHubRestCatalogTest.java

@@ -275,8 +275,8 @@ public void testListNamespaces_NestedNamespace() throws Exception {
     SearchResult mockResult = mock(SearchResult.class);
     List<SearchEntity> entitiesList =
         Arrays.asList(
-            createSearchEntity("urn:li:container:iceberg__parent.ns1"),
-            createSearchEntity("urn:li:container:iceberg__parent.ns2"));
+            createSearchEntity("urn:li:container:iceberg__platformInstance.parent.ns1"),
+            createSearchEntity("urn:li:container:iceberg__platformInstance.parent.ns2"));
     SearchEntityArray entities = new SearchEntityArray();
     entities.addAll(entitiesList);
     when(mockResult.getEntities()).thenReturn(entities);

metadata-service/iceberg-catalog/src/test/java/io/datahubproject/iceberg/catalog/UtilsTest.java

@@ -42,6 +42,10 @@ public void testContainerUrn() {
 
     assertNotNull(containerUrn);
     assertEquals(containerUrn.toString(), "urn:li:container:iceberg__testInstance.db.schema");
+
+    Namespace namespaceFromUrn =
+        Namespace.of(Utils.namespaceNameFromContainerUrn(containerUrn).split("\\."));
+    assertEquals(namespaceFromUrn, namespace);
   }
 
   @Test