Unblock test listNamespacesWithEmptyNamespace (#1289)

* Unblock test `listNamespacesWithEmptyNamespace`

* Use `containsExactly` to simplify the test

* Fix empty namespace behavior

* Address comments

* Block dropping empty namespace

* Improve error messages

Author: liamzwbao

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogTest.java

@@ -123,11 +123,11 @@
 import org.apache.polaris.service.types.NotificationType;
 import org.apache.polaris.service.types.TableUpdateNotification;
 import org.assertj.core.api.Assertions;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -164,6 +164,7 @@ public Map<String, String> getConfigOverrides() {
       new Schema(
           required(3, "id", Types.IntegerType.get(), "unique ID 🤪"),
           required(4, "data", Types.StringType.get()));
+  private static final String VIEW_QUERY = "select * from ns1.layer1_table";
   public static final String CATALOG_NAME = "polaris-catalog";
   public static final String TEST_ACCESS_KEY = "test_access_key";
   public static final String SECRET_ACCESS_KEY = "secret_access_key";
@@ -386,18 +387,48 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
     };
   }
 
-  /** TODO: Unblock this test, see: https://github.com/apache/polaris/issues/1272 */
-  @Override
   @Test
-  @Disabled(
-      """
-      Disabled because the behavior is not applicable to Polaris.
-      To unblock:
-      1) Align Polaris behavior with the superclass by handling empty namespaces the same way, or
-      2) Modify this test to expect an exception and add a Polaris-specific version.
-      """)
-  public void listNamespacesWithEmptyNamespace() {
-    super.listNamespacesWithEmptyNamespace();
+  public void testEmptyNamespace() {
+    IcebergCatalog catalog = catalog();
+    TableIdentifier tableInRootNs = TableIdentifier.of("table");
+    String expectedMessage = "Namespace does not exist: ''";
+
+    ThrowingCallable createEmptyNamespace = () -> catalog.createNamespace(Namespace.empty());
+    Assertions.assertThatThrownBy(createEmptyNamespace)
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Cannot create root namespace, as it already exists implicitly.");
+
+    ThrowingCallable dropEmptyNamespace = () -> catalog.dropNamespace(Namespace.empty());
+    Assertions.assertThatThrownBy(dropEmptyNamespace)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot drop root namespace");
+
+    ThrowingCallable createTable = () -> catalog.createTable(tableInRootNs, SCHEMA);
+    Assertions.assertThatThrownBy(createTable)
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessageContaining(expectedMessage);
+
+    ThrowingCallable createView =
+        () ->
+            catalog
+                .buildView(tableInRootNs)
+                .withSchema(SCHEMA)
+                .withDefaultNamespace(Namespace.empty())
+                .withQuery("spark", VIEW_QUERY)
+                .create();
+    Assertions.assertThatThrownBy(createView)
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessageContaining(expectedMessage);
+
+    ThrowingCallable listTables = () -> catalog.listTables(Namespace.empty());
+    Assertions.assertThatThrownBy(listTables)
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessageContaining(expectedMessage);
+
+    ThrowingCallable listViews = () -> catalog.listViews(Namespace.empty());
+    Assertions.assertThatThrownBy(listViews)
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessageContaining(expectedMessage);
   }
 
   @Test

service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -462,9 +462,9 @@ public boolean dropTable(TableIdentifier tableIdentifier, boolean purge) {
 
   @Override
   public List<TableIdentifier> listTables(Namespace namespace) {
-    if (!namespaceExists(namespace) && !namespace.isEmpty()) {
+    if (!namespaceExists(namespace)) {
       throw new NoSuchNamespaceException(
-          "Cannot list tables for namespace. Namespace does not exist: %s", namespace);
+          "Cannot list tables for namespace. Namespace does not exist: '%s'", namespace);
     }
 
     return listTableLike(PolarisEntitySubType.ICEBERG_TABLE, namespace);
@@ -633,11 +633,17 @@ private PolarisResolvedPathWrapper getResolvedParentNamespace(Namespace namespac
 
   @Override
   public boolean namespaceExists(Namespace namespace) {
-    return resolvedEntityView.getResolvedPath(namespace) != null;
+    return Optional.ofNullable(namespace)
+        .filter(ns -> !ns.isEmpty())
+        .map(resolvedEntityView::getResolvedPath)
+        .isPresent();
   }
 
   @Override
   public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyException {
+    if (namespace.isEmpty()) {
+      throw new IllegalArgumentException("Cannot drop root namespace");
+    }
     PolarisResolvedPathWrapper resolvedEntities = resolvedEntityView.getResolvedPath(namespace);
     if (resolvedEntities == null) {
       return false;
@@ -798,9 +804,9 @@ public void close() throws IOException {
 
   @Override
   public List<TableIdentifier> listViews(Namespace namespace) {
-    if (!namespaceExists(namespace) && !namespace.isEmpty()) {
+    if (!namespaceExists(namespace)) {
       throw new NoSuchNamespaceException(
-          "Cannot list views for namespace. Namespace does not exist: %s", namespace);
+          "Cannot list views for namespace. Namespace does not exist: '%s'", namespace);
     }
 
     return listTableLike(PolarisEntitySubType.ICEBERG_VIEW, namespace);
@@ -1251,7 +1257,7 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
       // TODO: Maybe avoid writing metadata if there's definitely a transaction conflict
       if (null == base && !namespaceExists(tableIdentifier.namespace())) {
         throw new NoSuchNamespaceException(
-            "Cannot create table %s. Namespace does not exist: %s",
+            "Cannot create table '%s'. Namespace does not exist: '%s'",
             tableIdentifier, tableIdentifier.namespace());
       }
 
@@ -1492,7 +1498,7 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) {
       LOGGER.debug("doCommit for view {} with base {}, metadata {}", identifier, base, metadata);
       if (null == base && !namespaceExists(identifier.namespace())) {
         throw new NoSuchNamespaceException(
-            "Cannot create view %s. Namespace does not exist: %s",
+            "Cannot create view '%s'. Namespace does not exist: '%s'",
             identifier, identifier.namespace());
       }