[Improvement]: Add TableFormat information to the return value of the Catalog listTable method.

ams/server/src/main/java/com/netease/arctic/server/IcebergRestCatalogService.java

@@ -250,7 +250,7 @@ public void listTablesInNamespace(Context ctx) {
     handleNamespace(ctx, (catalog, database) -> {
       checkDatabaseExist(catalog.exist(database), database);
       List<TableIdentifier> tableIdentifiers = catalog.listTables(database).stream()
-          .map(i -> TableIdentifier.of(database, i.getTableName()))
+          .map(i -> TableIdentifier.of(database, i.getIdentifier().getTableName()))
           .collect(Collectors.toList());
 
       return ListTablesResponse.builder()
@@ -289,7 +289,8 @@ public void createTable(Context ctx) {
           sortOrder != null ? sortOrder : SortOrder.unsorted(),
           location, request.properties()
       );
-      ServerTableIdentifier identifier = ServerTableIdentifier.of(catalog.name(), database, tableName);
+      ServerTableIdentifier identifier = ServerTableIdentifier.of(
+          catalog.name(), database, tableName, TableFormat.ICEBERG);
       String newMetadataFileLocation = IcebergTableUtil.genNewMetadataFileLocation(null, tableMetadata);
       FileIO io = newIcebergFileIo(catalog.getMetadata());
       try {

ams/server/src/main/java/com/netease/arctic/server/TableManagementService.java

@@ -25,6 +25,7 @@
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.OperationConflictException;
 import com.netease.arctic.ams.api.TableCommitMeta;
+import com.netease.arctic.ams.api.TableFormat;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.server.table.ServerTableIdentifier;
@@ -79,7 +80,8 @@ public void createTableMeta(TableMeta tableMeta) {
     if (tableMeta == null) {
       throw new IllegalArgumentException("table meta should not be null");
     }
-    ServerTableIdentifier identifier = ServerTableIdentifier.of(tableMeta.getTableIdentifier());
+    ServerTableIdentifier identifier = ServerTableIdentifier.of(
+        tableMeta.getTableIdentifier(), TableFormat.valueOf(tableMeta.getFormat()));
     CatalogMeta catalogMeta = getCatalog(identifier.getCatalog());
     TableMetadata tableMetadata = new TableMetadata(identifier, tableMeta, catalogMeta);
     tableService.createTable(tableMeta.tableIdentifier.getCatalog(), tableMetadata);

ams/server/src/main/java/com/netease/arctic/server/catalog/ExternalCatalog.java

@@ -1,6 +1,7 @@
 package com.netease.arctic.server.catalog;
 
 import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.ams.api.TableFormat;
 import com.netease.arctic.server.persistence.mapper.TableMetaMapper;
 import com.netease.arctic.server.table.ServerTableIdentifier;
 
@@ -10,9 +11,9 @@ protected ExternalCatalog(CatalogMeta metadata) {
     super(metadata);
   }
 
-  public void syncTable(String database, String tableName) {
+  public void syncTable(String database, String tableName, TableFormat format) {
     ServerTableIdentifier tableIdentifier =
-            ServerTableIdentifier.of(getMetadata().getCatalogName(), database, tableName);
+            ServerTableIdentifier.of(getMetadata().getCatalogName(), database, tableName, format);
     doAs(TableMetaMapper.class, mapper -> mapper.insertTable(tableIdentifier));
   }
 

ams/server/src/main/java/com/netease/arctic/server/catalog/IcebergCatalogImpl.java

@@ -1,11 +1,11 @@
 package com.netease.arctic.server.catalog;
 
 import com.netease.arctic.AmoroTable;
+import com.netease.arctic.TableIDWithFormat;
 import com.netease.arctic.ams.api.CatalogMeta;
-import com.netease.arctic.ams.api.TableIdentifier;
+import com.netease.arctic.ams.api.TableFormat;
 import com.netease.arctic.catalog.IcebergCatalogWrapper;
 import com.netease.arctic.formats.iceberg.IcebergTable;
-import com.netease.arctic.utils.CatalogUtil;
 import org.apache.iceberg.Table;
 
 import java.util.Collections;
@@ -43,17 +43,17 @@ public List<String> listDatabases() {
   }
 
   @Override
-  public List<TableIdentifier> listTables() {
-    return toAmsIdList(catalogWrapper.listTables());
+  public List<TableIDWithFormat> listTables() {
+    return catalogWrapper.listTables().stream()
+        .map(id -> TableIDWithFormat.of(id, TableFormat.ICEBERG))
+        .collect(Collectors.toList());
   }
 
   @Override
-  public List<TableIdentifier> listTables(String database) {
-    return toAmsIdList(catalogWrapper.listTables(database));
-  }
-
-  public List<TableIdentifier> toAmsIdList(List<com.netease.arctic.table.TableIdentifier> identifierList) {
-    return identifierList.stream().map(CatalogUtil::amsTaleId).collect(Collectors.toList());
+  public List<TableIDWithFormat> listTables(String database) {
+    return catalogWrapper.listTables(database).stream()
+        .map(id -> TableIDWithFormat.of(id, TableFormat.ICEBERG))
+        .collect(Collectors.toList());
   }
 
   @Override

ams/server/src/main/java/com/netease/arctic/server/catalog/InternalCatalog.java

@@ -1,5 +1,6 @@
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.TableIDWithFormat;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.server.exception.AlreadyExistsException;
@@ -60,22 +61,24 @@ public void dropDatabase(String databaseName) {
   }
 
   @Override
-  public List<TableIdentifier> listTables(String database) {
+  public List<TableIDWithFormat> listTables(String database) {
     return getAs(
         TableMetaMapper.class,
         mapper -> mapper.selectTableIdentifiersByDb(getMetadata().getCatalogName(), database))
         .stream()
-        .map(ServerTableIdentifier::getIdentifier)
+        .map(sid -> TableIDWithFormat.of(
+            com.netease.arctic.table.TableIdentifier.of(sid.getIdentifier()), sid.getFormat()))
         .collect(Collectors.toList());
   }
 
   @Override
-  public List<TableIdentifier> listTables() {
+  public List<TableIDWithFormat> listTables() {
     return getAs(
         TableMetaMapper.class,
         mapper -> mapper.selectTableIdentifiersByCatalog(getMetadata().getCatalogName()))
         .stream()
-        .map(ServerTableIdentifier::getIdentifier)
+        .map(sid -> TableIDWithFormat.of(
+            com.netease.arctic.table.TableIdentifier.of(sid.getIdentifier()), sid.getFormat()))
         .collect(Collectors.toList());
   }
 

ams/server/src/main/java/com/netease/arctic/server/catalog/MixedIcebergCatalogImpl.java

@@ -19,8 +19,9 @@
 package com.netease.arctic.server.catalog;
 
 import com.netease.arctic.AmoroTable;
+import com.netease.arctic.TableIDWithFormat;
 import com.netease.arctic.ams.api.CatalogMeta;
-import com.netease.arctic.ams.api.TableIdentifier;
+import com.netease.arctic.ams.api.TableFormat;
 import com.netease.arctic.formats.mixed.MixedIcebergTable;
 import com.netease.arctic.mixed.BasicMixedIcebergCatalog;
 import java.util.List;
@@ -56,18 +57,18 @@ public List<String> listDatabases() {
   }
 
   @Override
-  public List<TableIdentifier> listTables() {
+  public List<TableIDWithFormat> listTables() {
     return listDatabases().stream()
         .map(this::listTables)
         .flatMap(List::stream)
         .collect(Collectors.toList());
   }
 
   @Override
-  public List<TableIdentifier> listTables(String database) {
+  public List<TableIDWithFormat> listTables(String database) {
     return mixedIcebergCatalog.listTables(database)
         .stream()
-        .map(com.netease.arctic.table.TableIdentifier::buildTableIdentifier)
+        .map(id -> TableIDWithFormat.of(id, TableFormat.MIXED_ICEBERG))
         .collect(Collectors.toList());
   }
 

ams/server/src/main/java/com/netease/arctic/server/catalog/PaimonServerCatalog.java

@@ -20,11 +20,13 @@
 
 import com.netease.arctic.AmoroTable;
 import com.netease.arctic.CommonUnifiedCatalog;
+import com.netease.arctic.TableIDWithFormat;
 import com.netease.arctic.UnifiedCatalog;
 import com.netease.arctic.ams.api.CatalogMeta;
-import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.table.TableMetaStore;
 import com.netease.arctic.utils.CatalogUtil;
+
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
@@ -72,7 +74,7 @@ public List<String> listDatabases() {
   }
 
   @Override
-  public List<TableIdentifier> listTables() {
+  public List<TableIDWithFormat> listTables() {
     return doAs(() -> paimonCatalog.listDatabases()
         .stream()
         .map(this::listTables)
@@ -81,11 +83,8 @@ public List<TableIdentifier> listTables() {
   }
 
   @Override
-  public List<TableIdentifier> listTables(String database) {
-    return doAs(() -> paimonCatalog.listTableMetas(database)
-        .stream()
-        .map(t -> t.getIdentifier().buildTableIdentifier())
-        .collect(Collectors.toList()));
+  public List<TableIDWithFormat> listTables(String database) {
+    return doAs(() -> new ArrayList<>(paimonCatalog.listTables(database)));
   }
 
   @Override

ams/server/src/main/java/com/netease/arctic/server/catalog/ServerCatalog.java

@@ -1,8 +1,8 @@
 package com.netease.arctic.server.catalog;
 
 import com.netease.arctic.AmoroTable;
+import com.netease.arctic.TableIDWithFormat;
 import com.netease.arctic.ams.api.CatalogMeta;
-import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.server.persistence.PersistentBase;
 import com.netease.arctic.server.persistence.mapper.CatalogMetaMapper;
 
@@ -36,9 +36,9 @@ public void updateMetadata(CatalogMeta metadata) {
 
   public abstract List<String> listDatabases();
 
-  public abstract List<TableIdentifier> listTables();
+  public abstract List<TableIDWithFormat> listTables();
 
-  public abstract List<TableIdentifier> listTables(String database);
+  public abstract List<TableIDWithFormat> listTables(String database);
 
   public abstract AmoroTable<?> loadTable(String database, String tableName);
 }

ams/server/src/main/java/com/netease/arctic/server/dashboard/ServerTableDescriptor.java

@@ -1,7 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.server.dashboard;
 
 import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.TableFormat;
+import com.netease.arctic.ams.api.TableIdentifier;
+import com.netease.arctic.server.catalog.ServerCatalog;
 import com.netease.arctic.server.dashboard.model.DDLInfo;
 import com.netease.arctic.server.dashboard.model.PartitionBaseInfo;
 import com.netease.arctic.server.dashboard.model.PartitionFileBaseInfo;
@@ -11,7 +31,6 @@
 import com.netease.arctic.server.optimizing.OptimizingTaskMeta;
 import com.netease.arctic.server.persistence.PersistentBase;
 import com.netease.arctic.server.persistence.mapper.OptimizingMapper;
-import com.netease.arctic.server.table.ServerTableIdentifier;
 import com.netease.arctic.server.table.TableService;
 import org.apache.commons.collections.CollectionUtils;
 
@@ -40,38 +59,38 @@ public ServerTableDescriptor(TableService tableService) {
     }
   }
 
-  public ServerTableMeta getTableDetail(ServerTableIdentifier tableIdentifier) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public ServerTableMeta getTableDetail(TableIdentifier tableIdentifier) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTableDetail(amoroTable);
   }
 
-  public List<TransactionsOfTable> getTransactions(ServerTableIdentifier tableIdentifier) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public List<TransactionsOfTable> getTransactions(TableIdentifier tableIdentifier) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTransactions(amoroTable);
   }
 
-  public List<PartitionFileBaseInfo> getTransactionDetail(ServerTableIdentifier tableIdentifier, long transactionId) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public List<PartitionFileBaseInfo> getTransactionDetail(TableIdentifier tableIdentifier, long transactionId) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTransactionDetail(amoroTable, transactionId);
   }
 
-  public List<DDLInfo> getTableOperations(ServerTableIdentifier tableIdentifier) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public List<DDLInfo> getTableOperations(TableIdentifier tableIdentifier) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTableOperations(amoroTable);
   }
 
-  public List<PartitionBaseInfo> getTablePartition(ServerTableIdentifier tableIdentifier) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public List<PartitionBaseInfo> getTablePartition(TableIdentifier tableIdentifier) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTablePartitions(amoroTable);
   }
 
-  public List<PartitionFileBaseInfo> getTableFile(ServerTableIdentifier tableIdentifier, String partition) {
-    AmoroTable<?> amoroTable = tableService.loadTable(tableIdentifier);
+  public List<PartitionFileBaseInfo> getTableFile(TableIdentifier tableIdentifier, String partition) {
+    AmoroTable<?> amoroTable = loadTable(tableIdentifier);
     FormatTableDescriptor formatTableDescriptor = formatDescriptorMap.get(amoroTable.format());
     return formatTableDescriptor.getTableFiles(amoroTable, partition);
   }
@@ -92,4 +111,9 @@ public List<OptimizingTaskMeta> getOptimizingTasks(List<OptimizingProcessMeta> p
         OptimizingMapper.class,
         mapper -> mapper.selectOptimizeTaskMetas(processIds));
   }
+
+  private AmoroTable<?> loadTable(TableIdentifier identifier) {
+    ServerCatalog catalog = tableService.getServerCatalog(identifier.getCatalog());
+    return catalog.loadTable(identifier.getDatabase(), identifier.getTableName());
+  }
 }