[AMORO-1906]Refactor code of "Table" and "Catalog" to support Paimon format integration

ams/api/src/main/java/com/netease/arctic/ams/api/TableFormat.java

@@ -28,4 +28,13 @@ public enum TableFormat {
   MIXED_ICEBERG,
   MIXED_HIVE,
   PAIMON;
+
+  public boolean in(TableFormat... tableFormats) {
+    for (TableFormat tableFormat : tableFormats) {
+      if (this == tableFormat) {
+        return true;
+      }
+    }
+    return false;
+  }
 }

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

@@ -19,6 +19,7 @@
 package com.netease.arctic.server;
 
 import com.google.common.base.Preconditions;
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.OptimizerRegisterInfo;
 import com.netease.arctic.ams.api.OptimizingService;
@@ -45,7 +46,6 @@
 import com.netease.arctic.server.table.TableRuntime;
 import com.netease.arctic.server.table.TableRuntimeMeta;
 import com.netease.arctic.server.utils.Configurations;
-import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableProperties;
 import org.apache.commons.collections.CollectionUtils;
 import org.slf4j.Logger;
@@ -332,7 +332,7 @@ public void handleConfigChanged(TableRuntime tableRuntime, TableConfiguration or
     }
 
     @Override
-    public void handleTableAdded(ArcticTable table, TableRuntime tableRuntime) {
+    public void handleTableAdded(AmoroTable<?> table, TableRuntime tableRuntime) {
       getOptionalQueueByGroup(tableRuntime.getOptimizerGroup()).ifPresent(q -> q.refreshTable(tableRuntime));
     }
 

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

@@ -22,6 +22,10 @@ public static ServerCatalog buildServerCatalog(CatalogMeta catalogMeta, Configur
     Set<TableFormat> tableFormats = CatalogUtil.tableFormats(catalogMeta);
     TableFormat tableFormat = tableFormats.iterator().next();
 
+    if (tableFormat == TableFormat.PAIMON) {
+      return new PaimonServerCatalog(catalogMeta);
+    }
+
     switch (type) {
       case CATALOG_TYPE_HADOOP:
         if (TableFormat.ICEBERG == tableFormat) {

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

@@ -1,10 +1,12 @@
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.catalog.IcebergCatalogWrapper;
-import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.formats.iceberg.IcebergTable;
 import com.netease.arctic.utils.CatalogUtil;
+import org.apache.iceberg.Table;
 
 import java.util.Collections;
 import java.util.List;
@@ -55,8 +57,11 @@ public List<TableIdentifier> toAmsIdList(List<com.netease.arctic.table.TableIden
   }
 
   @Override
-  public ArcticTable loadTable(String database, String tableName) {
-    return catalogWrapper.loadTable(com.netease.arctic.table.TableIdentifier.of(catalogWrapper.name(), database,
-        tableName));
+  public AmoroTable<Table> loadTable(String database, String tableName) {
+    com.netease.arctic.table.TableIdentifier identifier = com.netease.arctic.table.TableIdentifier.of(
+        catalogWrapper.name(),
+        database,
+        tableName);
+    return new IcebergTable(identifier, catalogWrapper.loadTable(identifier).asUnkeyedTable());
   }
 }

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

@@ -1,7 +1,9 @@
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.catalog.IcebergCatalogWrapper;
+import com.netease.arctic.formats.iceberg.IcebergTable;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.ArcticFileIOAdapter;
 import com.netease.arctic.server.ArcticManagementConf;
@@ -11,7 +13,6 @@
 import com.netease.arctic.server.table.TableMetadata;
 import com.netease.arctic.server.utils.Configurations;
 import com.netease.arctic.server.utils.IcebergTableUtil;
-import com.netease.arctic.table.ArcticTable;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.TableOperations;
@@ -54,7 +55,7 @@ public void updateMetadata(CatalogMeta metadata) {
   }
 
   @Override
-  public ArcticTable loadTable(String database, String tableName) {
+  public AmoroTable<?> loadTable(String database, String tableName) {
     TableMetadata tableMetadata = getAs(TableMetaMapper.class, mapper ->
         mapper.selectTableMetaByName(getMetadata().getCatalogName(), database, tableName));
     if (tableMetadata == null) {
@@ -64,9 +65,11 @@ public ArcticTable loadTable(String database, String tableName) {
     ArcticFileIO fileIO = new ArcticFileIOAdapter(io);
     TableOperations ops = InternalTableOperations.buildForLoad(tableMetadata, io);
     BaseTable table = new BaseTable(ops, TableIdentifier.of(database, tableName).toString());
-    return new IcebergCatalogWrapper.BasicIcebergTable(
-        com.netease.arctic.table.TableIdentifier.of(name(), database, tableName),
-        table, fileIO, getMetadata().getCatalogProperties()
+    com.netease.arctic.table.TableIdentifier tableIdentifier =
+        com.netease.arctic.table.TableIdentifier.of(name(), database, tableName);
+    return new IcebergTable(tableIdentifier,
+        new IcebergCatalogWrapper.BasicIcebergTable(
+            tableIdentifier, table, fileIO, getMetadata().getCatalogProperties())
     );
   }
 

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

@@ -1,14 +1,15 @@
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.catalog.MixedTables;
+import com.netease.arctic.formats.mixed.MixedIcebergTable;
 import com.netease.arctic.server.persistence.mapper.TableMetaMapper;
 import com.netease.arctic.server.table.TableMetadata;
-import com.netease.arctic.table.ArcticTable;
 
 public class InternalMixedCatalogImpl extends InternalCatalog {
 
-  private final MixedTables tables;
+  protected final MixedTables tables;
 
   protected InternalMixedCatalogImpl(CatalogMeta metadata) {
     super(metadata);
@@ -27,10 +28,13 @@
   }
 
   @Override
-  public ArcticTable loadTable(String database, String tableName) {
+  public AmoroTable<?> loadTable(String database, String tableName) {
     TableMetadata tableMetadata = getAs(TableMetaMapper.class, mapper ->
         mapper.selectTableMetaByName(getMetadata().getCatalogName(), database, tableName));
-    return tableMetadata == null ? null : tables.loadTableByMeta(tableMetadata.buildTableMeta());
+    if (tableMetadata == null) {
+      return null;
+    }
+    return new MixedIcebergTable(tables.loadTableByMeta(tableMetadata.buildTableMeta()));
   }
 
   protected MixedTables tables() {

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

@@ -18,10 +18,14 @@
 
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.formats.mixed.MixedHiveTable;
 import com.netease.arctic.hive.CachedHiveClientPool;
 import com.netease.arctic.hive.HMSClient;
 import com.netease.arctic.hive.catalog.MixedHiveTables;
+import com.netease.arctic.server.persistence.mapper.TableMetaMapper;
+import com.netease.arctic.server.table.TableMetadata;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.thrift.TException;
 
@@ -42,6 +46,16 @@
     hiveClientPool = ((MixedHiveTables) tables()).getHiveClientPool();
   }
 
+  @Override
+  public AmoroTable<?> loadTable(String database, String tableName) {
+    TableMetadata tableMetadata = getAs(TableMetaMapper.class, mapper ->
+        mapper.selectTableMetaByName(getMetadata().getCatalogName(), database, tableName));
+    if (tableMetadata == null) {
+      return null;
+    }
+    return new MixedHiveTable(tables.loadTableByMeta(tableMetadata.buildTableMeta()));
+  }
+
   @Override
   public void createDatabase(String databaseName) {
     // do not handle database operations

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

@@ -18,11 +18,11 @@
 
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
+import com.netease.arctic.formats.mixed.MixedIcebergTable;
 import com.netease.arctic.mixed.BasicMixedIcebergCatalog;
-import com.netease.arctic.table.ArcticTable;
-
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -72,7 +72,8 @@ public List<TableIdentifier> listTables(String database) {
   }
 
   @Override
-  public ArcticTable loadTable(String database, String tableName) {
-    return mixedIcebergCatalog.loadTable(com.netease.arctic.table.TableIdentifier.of(name(), database, tableName));
+  public AmoroTable<?> loadTable(String database, String tableName) {
+    return new MixedIcebergTable(mixedIcebergCatalog.loadTable(
+        com.netease.arctic.table.TableIdentifier.of(name(), database, tableName)));
   }
 }

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

@@ -0,0 +1,99 @@
+/*
+ * 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.catalog;
+
+import com.netease.arctic.AmoroTable;
+import com.netease.arctic.CommonUnifiedCatalog;
+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.List;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+
+public class PaimonServerCatalog extends ExternalCatalog {
+
+  private volatile UnifiedCatalog paimonCatalog;
+
+  private volatile TableMetaStore tableMetaStore;
+
+  protected PaimonServerCatalog(CatalogMeta metadata) {
+    super(metadata);
+    this.tableMetaStore = CatalogUtil.buildMetaStore(metadata);
+    this.paimonCatalog = doAs(() -> new CommonUnifiedCatalog(
+        null,
+        metadata,
+        metadata.catalogProperties
+    ));
+  }
+
+  @Override
+  public void updateMetadata(CatalogMeta metadata) {
+    super.updateMetadata(metadata);
+    this.tableMetaStore = CatalogUtil.buildMetaStore(metadata);
+    this.paimonCatalog = doAs(() -> new CommonUnifiedCatalog(
+        null,
+        metadata,
+        metadata.catalogProperties
+    ));
+  }
+
+  @Override
+  public boolean exist(String database) {
+    return doAs(() -> paimonCatalog.exist(database));
+  }
+
+  @Override
+  public boolean exist(String database, String tableName) {
+    return doAs(() -> paimonCatalog.exist(database, tableName));
+  }
+
+  @Override
+  public List<String> listDatabases() {
+    return doAs(() -> paimonCatalog.listDatabases());
+  }
+
+  @Override
+  public List<TableIdentifier> listTables() {
+    return doAs(() -> paimonCatalog.listDatabases()
+        .stream()
+        .map(this::listTables)
+        .flatMap(List::stream)
+        .collect(Collectors.toList()));
+  }
+
+  @Override
+  public List<TableIdentifier> listTables(String database) {
+    return doAs(() -> paimonCatalog.listTableMetas(database)
+        .stream()
+        .map(t -> t.getIdentifier().buildTableIdentifier())
+        .collect(Collectors.toList()));
+  }
+
+  @Override
+  public AmoroTable<?> loadTable(String database, String tableName) {
+    return doAs(() -> paimonCatalog.loadTable(database, tableName));
+  }
+
+  private <T> T doAs(Callable<T> callable) {
+    return tableMetaStore.doAs(callable);
+  }
+}

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

@@ -1,10 +1,10 @@
 package com.netease.arctic.server.catalog;
 
+import com.netease.arctic.AmoroTable;
 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;
-import com.netease.arctic.table.ArcticTable;
 
 import java.util.List;
 
@@ -40,5 +40,5 @@ public void updateMetadata(CatalogMeta metadata) {
 
   public abstract List<TableIdentifier> listTables(String database);
 
-  public abstract ArcticTable loadTable(String database, String tableName);
+  public abstract AmoroTable<?> loadTable(String database, String tableName);
 }