Core: Add tests for catalogs supporting empty namespaces (#9890)

core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java

@@ -146,7 +146,7 @@ public List<TableIdentifier> listTables(Namespace namespace) {
     }
 
     return tables.keySet().stream()
-        .filter(t -> namespace.isEmpty() || t.namespace().equals(namespace))
+        .filter(t -> t.namespace().equals(namespace))
         .sorted(Comparator.comparing(TableIdentifier::toString))
         .collect(Collectors.toList());
   }
@@ -290,6 +290,7 @@ public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespac
 
     List<Namespace> filteredNamespaces =
         namespaces.keySet().stream()
+            .filter(n -> !n.isEmpty())
             .filter(n -> DOT.join(n.levels()).startsWith(searchNamespaceString))
             .collect(Collectors.toList());
 
@@ -323,7 +324,7 @@ public List<TableIdentifier> listViews(Namespace namespace) {
     }
 
     return views.keySet().stream()
-        .filter(v -> namespace.isEmpty() || v.namespace().equals(namespace))
+        .filter(v -> v.namespace().equals(namespace))
         .sorted(Comparator.comparing(TableIdentifier::toString))
         .collect(Collectors.toList());
   }

core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java

@@ -500,6 +500,10 @@ private JdbcUtil() {}
 
   static Namespace stringToNamespace(String namespace) {
     Preconditions.checkArgument(namespace != null, "Invalid namespace %s", namespace);
+    if (namespace.isEmpty()) {
+      return Namespace.empty();
+    }
+
     return Namespace.of(Iterables.toArray(SPLITTER_DOT.split(namespace), String.class));
   }
 

core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java

@@ -22,6 +22,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.setMaxStackTraceElementsDisplayed;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -183,6 +184,10 @@ protected boolean supportsNamesWithDot() {
     return true;
   }
 
+  protected boolean supportsEmptyNamespace() {
+    return false;
+  }
+
   @Test
   public void testCreateNamespace() {
     C catalog = catalog();
@@ -978,6 +983,74 @@ public void testListTables() {
     assertEmpty("Should not contain ns_2.table_1 after drop", catalog, ns2);
   }
 
+  @Test
+  public void listNamespacesWithEmptyNamespace() {
+    catalog().createNamespace(NS);
+
+    assertThat(catalog().namespaceExists(Namespace.empty())).isFalse();
+    assertThat(catalog().listNamespaces()).contains(NS).doesNotContain(Namespace.empty());
+    assertThat(catalog().listNamespaces(Namespace.empty()))
+        .contains(NS)
+        .doesNotContain(Namespace.empty());
+  }
+
+  @Test
+  public void createAndDropEmptyNamespace() {
+    assumeThat(supportsEmptyNamespace())
+        .as("Only valid for catalogs that support creating/dropping empty namespaces")
+        .isTrue();
+
+    assertThat(catalog().namespaceExists(Namespace.empty())).isFalse();
+    catalog().createNamespace(Namespace.empty());
+    assertThat(catalog().namespaceExists(Namespace.empty())).isTrue();
+
+    // TODO: if a catalog supports creating an empty namespace, what should be the expected behavior
+    // when listing all namespaces?
+    assertThat(catalog().listNamespaces()).isEmpty();
+    assertThat(catalog().listNamespaces(Namespace.empty())).isEmpty();
+
+    catalog().dropNamespace(Namespace.empty());
+    assertThat(catalog().namespaceExists(Namespace.empty())).isFalse();
+  }
+
+  @Test
+  public void namespacePropertiesOnEmptyNamespace() {
+    assumeThat(supportsEmptyNamespace())
+        .as("Only valid for catalogs that support properties on empty namespaces")
+        .isTrue();
+
+    catalog().createNamespace(Namespace.empty());
+
+    Map<String, String> properties = ImmutableMap.of("owner", "user", "created-at", "sometime");
+    catalog().setProperties(Namespace.empty(), properties);
+
+    assertThat(catalog().loadNamespaceMetadata(Namespace.empty())).containsAllEntriesOf(properties);
+
+    catalog().removeProperties(Namespace.empty(), ImmutableSet.of("owner"));
+    assertThat(catalog().loadNamespaceMetadata(Namespace.empty()))
+        .containsAllEntriesOf(ImmutableMap.of("created-at", "sometime"));
+  }
+
+  @Test
+  public void listTablesInEmptyNamespace() {
+    assumeThat(supportsEmptyNamespace())
+        .as("Only valid for catalogs that support listing tables in empty namespaces")
+        .isTrue();
+
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(Namespace.empty());
+      catalog().createNamespace(NS);
+    }
+
+    TableIdentifier table1 = TableIdentifier.of(Namespace.empty(), "table_1");
+    TableIdentifier table2 = TableIdentifier.of(NS, "table_2");
+
+    catalog().buildTable(table1, SCHEMA).create();
+    catalog().buildTable(table2, SCHEMA).create();
+
+    assertThat(catalog().listTables(Namespace.empty())).containsExactly(table1);
+  }
+
   @Test
   public void testUpdateTableSchema() {
     C catalog = catalog();

core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryCatalog.java

@@ -48,4 +48,9 @@ protected InMemoryCatalog initCatalog(
   protected boolean requiresNamespaceCreate() {
     return true;
   }
+
+  @Override
+  protected boolean supportsEmptyNamespace() {
+    return true;
+  }
 }

core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryViewCatalog.java

@@ -52,4 +52,9 @@ protected Catalog tableCatalog() {
   protected boolean requiresNamespaceCreate() {
     return true;
   }
+
+  @Override
+  protected boolean supportsEmptyNamespace() {
+    return true;
+  }
 }

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java

@@ -103,12 +103,12 @@ protected JdbcCatalog catalog() {
   }
 
   @Override
-  protected boolean supportsNamespaceProperties() {
+  protected boolean supportsNestedNamespaces() {
     return true;
   }
 
   @Override
-  protected boolean supportsNestedNamespaces() {
+  protected boolean supportsEmptyNamespace() {
     return true;
   }
 
@@ -763,11 +763,11 @@ public void testListNamespace() {
 
     List<Namespace> nsp3 = catalog.listNamespaces();
     Set<String> tblSet2 = Sets.newHashSet(nsp3.stream().map(Namespace::toString).iterator());
-    assertThat(tblSet2).hasSize(5).contains("db", "db2", "d_", "d%", "");
+    assertThat(tblSet2).hasSize(4).contains("db", "db2", "d_", "d%");
 
     List<Namespace> nsp4 = catalog.listNamespaces();
     Set<String> tblSet3 = Sets.newHashSet(nsp4.stream().map(Namespace::toString).iterator());
-    assertThat(tblSet3).hasSize(5).contains("db", "db2", "d_", "d%", "");
+    assertThat(tblSet3).hasSize(4).contains("db", "db2", "d_", "d%");
 
     List<Namespace> nsp5 = catalog.listNamespaces(Namespace.of("d_"));
     assertThat(nsp5).hasSize(1);

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcUtil.java

@@ -143,4 +143,10 @@ public void testV0toV1SqlStatements() throws Exception {
       assertThat(updated).isEqualTo(1);
     }
   }
+
+  @Test
+  public void emptyNamespaceInIdentifier() {
+    assertThat(JdbcUtil.stringToTableIdentifier("", "tblName"))
+        .isEqualTo(TableIdentifier.of(Namespace.empty(), "tblName"));
+  }
 }

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcViewCatalog.java

@@ -80,6 +80,11 @@ protected boolean requiresNamespaceCreate() {
     return true;
   }
 
+  @Override
+  protected boolean supportsEmptyNamespace() {
+    return true;
+  }
+
   @Test
   public void testCommitExceptionWithoutMessage() {
     TableIdentifier identifier = TableIdentifier.of("namespace1", "view");

core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java

@@ -21,6 +21,7 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -73,6 +74,10 @@ protected boolean supportsServerSideRetry() {
     return false;
   }
 
+  protected boolean supportsEmptyNamespace() {
+    return false;
+  }
+
   @Test
   public void basicCreateView() {
     TableIdentifier identifier = TableIdentifier.of("ns", "view");
@@ -814,6 +819,39 @@ public void listViews() {
     assertThat(catalog().listViews(ns2)).isEmpty();
   }
 
+  @Test
+  public void listViewsInEmptyNamespace() {
+    assumeThat(supportsEmptyNamespace())
+        .as("Only valid for catalogs that support listing views in empty namespaces")
+        .isTrue();
+
+    Namespace namespace = Namespace.of("ns");
+
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(Namespace.empty());
+      catalog().createNamespace(namespace);
+    }
+
+    TableIdentifier view1 = TableIdentifier.of(Namespace.empty(), "view1");
+    TableIdentifier view2 = TableIdentifier.of(namespace, "view2");
+
+    catalog()
+        .buildView(view1)
+        .withSchema(SCHEMA)
+        .withDefaultNamespace(view1.namespace())
+        .withQuery("spark", "select * from ns.tbl")
+        .create();
+
+    catalog()
+        .buildView(view2)
+        .withSchema(SCHEMA)
+        .withDefaultNamespace(view2.namespace())
+        .withQuery("spark", "select * from ns.tbl")
+        .create();
+
+    assertThat(catalog().listViews(Namespace.empty())).containsExactly(view1);
+  }
+
   @Test
   public void listViewsAndTables() {
     Assumptions.assumeThat(tableCatalog())

spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java

@@ -28,6 +28,7 @@
 import java.util.Random;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.IcebergBuild;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.catalog.Catalog;
@@ -1532,39 +1533,37 @@ public void showViews() throws NoSuchTableException {
 
     // spark stores temp views case-insensitive by default
     Object[] tempView = row("", tempViewForListing.toLowerCase(Locale.ROOT), true);
-    assertThat(sql("SHOW VIEWS"))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false),
-            row(NAMESPACE.toString(), prefixV3, false),
-            row(NAMESPACE.toString(), v1, false),
-            tempView);
-
-    assertThat(sql("SHOW VIEWS IN %s", catalogName))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false),
-            row(NAMESPACE.toString(), prefixV3, false),
-            row(NAMESPACE.toString(), v1, false),
-            tempView);
+    Object[] v1Row = row(NAMESPACE.toString(), v1, false);
+    Object[] v2Row = row(NAMESPACE.toString(), prefixV2, false);
+    Object[] v3Row = row(NAMESPACE.toString(), prefixV3, false);
+    assertThat(sql("SHOW VIEWS")).contains(v2Row, v3Row, v1Row, tempView);
+
+    if (!"rest".equals(catalogConfig.get(CatalogUtil.ICEBERG_CATALOG_TYPE))) {
+      // REST catalog requires a namespace
+      assertThat(sql("SHOW VIEWS IN %s", catalogName))
+          .contains(tempView)
+          .doesNotContain(v1Row, v2Row, v3Row);
+    }
 
     assertThat(sql("SHOW VIEWS IN %s.%s", catalogName, NAMESPACE))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false),
-            row(NAMESPACE.toString(), prefixV3, false),
-            row(NAMESPACE.toString(), v1, false),
-            tempView);
+        .contains(v2Row, v3Row, v1Row, tempView);
 
     assertThat(sql("SHOW VIEWS LIKE 'pref*'"))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false), row(NAMESPACE.toString(), prefixV3, false));
+        .contains(v2Row, v3Row)
+        .doesNotContain(v1Row, tempView);
 
     assertThat(sql("SHOW VIEWS LIKE 'non-existing'")).isEmpty();
 
-    assertThat(sql("SHOW VIEWS IN spark_catalog.default")).contains(tempView);
+    sql("CREATE NAMESPACE IF NOT EXISTS spark_catalog.%s", NAMESPACE);
+    assertThat(sql("SHOW VIEWS IN spark_catalog.%s", NAMESPACE))
+        .contains(tempView)
+        .doesNotContain(v1Row, v2Row, v3Row);
 
     assertThat(sql("SHOW VIEWS IN global_temp"))
         .contains(
             // spark stores temp views case-insensitive by default
-            row("global_temp", globalViewForListing.toLowerCase(Locale.ROOT), true), tempView);
+            row("global_temp", globalViewForListing.toLowerCase(Locale.ROOT), true), tempView)
+        .doesNotContain(v1Row, v2Row, v3Row);
 
     sql("USE spark_catalog");
     assertThat(sql("SHOW VIEWS")).contains(tempView);

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java

@@ -1561,45 +1561,39 @@ public void showViews() throws NoSuchTableException {
 
     // spark stores temp views case-insensitive by default
     Object[] tempView = row("", tempViewForListing.toLowerCase(Locale.ROOT), true);
-    assertThat(sql("SHOW VIEWS"))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false),
-            row(NAMESPACE.toString(), prefixV3, false),
-            row(NAMESPACE.toString(), v1, false),
-            tempView);
+    Object[] v1Row = row(NAMESPACE.toString(), v1, false);
+    Object[] v2Row = row(NAMESPACE.toString(), prefixV2, false);
+    Object[] v3Row = row(NAMESPACE.toString(), prefixV3, false);
+    assertThat(sql("SHOW VIEWS")).contains(v2Row, v3Row, v1Row, tempView);
 
     if (!"rest".equals(catalogConfig.get(CatalogUtil.ICEBERG_CATALOG_TYPE))) {
       // REST catalog requires a namespace
       assertThat(sql("SHOW VIEWS IN %s", catalogName))
-          .contains(
-              row(NAMESPACE.toString(), prefixV2, false),
-              row(NAMESPACE.toString(), prefixV3, false),
-              row(NAMESPACE.toString(), v1, false),
-              tempView);
+          .contains(tempView)
+          .doesNotContain(v1Row, v2Row, v3Row);
     }
 
     assertThat(sql("SHOW VIEWS IN %s.%s", catalogName, NAMESPACE))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false),
-            row(NAMESPACE.toString(), prefixV3, false),
-            row(NAMESPACE.toString(), v1, false),
-            tempView);
+        .contains(v2Row, v3Row, v1Row, tempView);
 
     assertThat(sql("SHOW VIEWS LIKE 'pref*'"))
-        .contains(
-            row(NAMESPACE.toString(), prefixV2, false), row(NAMESPACE.toString(), prefixV3, false));
+        .contains(v2Row, v3Row)
+        .doesNotContain(v1Row, tempView);
 
     assertThat(sql("SHOW VIEWS LIKE 'non-existing'")).isEmpty();
 
     if (!catalogName.equals(SPARK_CATALOG)) {
       sql("CREATE NAMESPACE IF NOT EXISTS spark_catalog.%s", NAMESPACE);
-      assertThat(sql("SHOW VIEWS IN spark_catalog.%s", NAMESPACE)).contains(tempView);
+      assertThat(sql("SHOW VIEWS IN spark_catalog.%s", NAMESPACE))
+          .contains(tempView)
+          .doesNotContain(v1Row, v2Row, v3Row);
     }
 
     assertThat(sql("SHOW VIEWS IN global_temp"))
         .contains(
             // spark stores temp views case-insensitive by default
-            row("global_temp", globalViewForListing.toLowerCase(Locale.ROOT), true), tempView);
+            row("global_temp", globalViewForListing.toLowerCase(Locale.ROOT), true), tempView)
+        .doesNotContain(v1Row, v2Row, v3Row);
 
     sql("USE spark_catalog");
     assertThat(sql("SHOW VIEWS")).contains(tempView);