[Iceberg SQL] Create cached tables with correct name

sdks/java/extensions/sql/iceberg/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergMetastore.java

@@ -94,7 +94,7 @@ public void dropTable(String tableName) {
   @Override
   public Map<String, Table> getTables() {
     for (String id : catalogConfig.listTables(database)) {
-      String name = TableName.create(id).getTableName();
+      String name = tableName(id);
       @Nullable Table cachedTable = cachedTables.get(name);
       if (cachedTable == null) {
         Table table = checkStateNotNull(loadTable(id));
@@ -116,6 +116,10 @@ public Map<String, Table> getTables() {
     return table;
   }
 
+  private String tableName(String tableId) {
+    return TableName.create(tableId).getTableName();
+  }
+
   private String getIdentifier(String name) {
     return database + "." + name;
   }
@@ -133,7 +137,7 @@ private String getIdentifier(Table table) {
     }
     return Table.builder()
         .type(getTableType())
-        .name(identifier)
+        .name(tableName(identifier))
         .schema(tableInfo.getSchema())
         .properties(TableUtils.parseProperties(tableInfo.getProperties()))
         .build();

sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergMetastoreTest.java

@@ -23,6 +23,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Map;
 import java.util.UUID;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
@@ -89,6 +90,18 @@ public void testGetTables() {
     assertEquals(ImmutableSet.of("my_table_1", "my_table_2"), metastore().getTables().keySet());
   }
 
+  @Test
+  public void testGetTablesLoadsUncachedTablesWithSimpleNames() {
+    String tableName = "uncached_table";
+    catalog.catalogConfig.createTable(
+        catalog.currentDatabase() + "." + tableName, Schema.of(), null, null);
+
+    Map<String, Table> tables = metastore().getTables();
+
+    assertEquals(ImmutableSet.of(tableName), tables.keySet());
+    assertEquals(tableName, tables.get(tableName).getName());
+  }
+
   @Test
   public void testSupportsPartitioning() {
     Table table = Table.builder().name("my_table_1").schema(Schema.of()).type("iceberg").build();

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergCatalogConfig.java

@@ -276,7 +276,7 @@ public boolean dropTable(String tableIdentifier) {
 
   public Set<String> listTables(String namespace) {
     return catalog().listTables(Namespace.of(namespace)).stream()
-        .map(TableIdentifier::name)
+        .map(TableIdentifier::toString)
         .collect(Collectors.toSet());
   }