[Improvement]: Remove AmsClient in UnkeyedTable and KeyedTable

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

@@ -2,6 +2,7 @@
 
 import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.data.FileNameRules;
+import com.netease.arctic.op.SnapshotSummary;
 import com.netease.arctic.server.dashboard.model.DDLInfo;
 import com.netease.arctic.server.dashboard.model.PartitionBaseInfo;
 import com.netease.arctic.server.dashboard.model.PartitionFileBaseInfo;
@@ -14,7 +15,6 @@
 import com.netease.arctic.server.table.ServerTableIdentifier;
 import com.netease.arctic.server.table.TableService;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.trace.SnapshotSummary;
 import com.netease.arctic.utils.ManifestEntryFields;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.iceberg.DataFile;
@@ -36,7 +36,6 @@
 import org.apache.iceberg.util.PropertyUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -229,7 +228,8 @@
   }
 
   public List<OptimizingTaskMeta> getOptimizingTasks(long processId) {
-    return getAs(OptimizingMapper.class,
+    return getAs(
+        OptimizingMapper.class,
         mapper -> mapper.selectOptimizeTaskMetas(Collections.singletonList(processId)));
   }
 
@@ -239,7 +239,8 @@
     }
     List<Long> processIds = processMetaList.stream()
         .map(OptimizingProcessMeta::getProcessId).collect(Collectors.toList());
-    return getAs(OptimizingMapper.class,
+    return getAs(
+        OptimizingMapper.class,
         mapper -> mapper.selectOptimizeTaskMetas(processIds));
   }
 

ams/server/src/main/java/com/netease/arctic/server/optimizing/KeyedTableCommit.java

@@ -5,11 +5,11 @@
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.op.OverwriteBaseFiles;
+import com.netease.arctic.op.SnapshotSummary;
 import com.netease.arctic.optimizing.RewriteFilesInput;
 import com.netease.arctic.optimizing.RewriteFilesOutput;
 import com.netease.arctic.server.exception.OptimizingCommitException;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.trace.SnapshotSummary;
 import com.netease.arctic.utils.ContentFiles;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
@@ -23,13 +23,11 @@
 import org.glassfish.jersey.internal.guava.Sets;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
-
 import static com.netease.arctic.hive.op.UpdateHiveFiles.DELETE_UNTRACKED_HIVE_FILE;
 import static com.netease.arctic.server.ArcticServiceConstants.INVALID_SNAPSHOT_ID;
 

ams/server/src/main/java/com/netease/arctic/server/optimizing/UnKeyedTableCommit.java

@@ -25,14 +25,14 @@
 import com.netease.arctic.hive.utils.HivePartitionUtil;
 import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.hive.utils.TableTypeUtil;
+import com.netease.arctic.op.SnapshotSummary;
 import com.netease.arctic.optimizing.OptimizingInputProperties;
 import com.netease.arctic.optimizing.RewriteFilesOutput;
 import com.netease.arctic.server.ArcticServiceConstants;
 import com.netease.arctic.server.exception.OptimizingCommitException;
 import com.netease.arctic.server.utils.IcebergTableUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.UnkeyedTable;
-import com.netease.arctic.trace.SnapshotSummary;
 import com.netease.arctic.utils.ContentFiles;
 import com.netease.arctic.utils.TableFileUtil;
 import org.apache.commons.collections.CollectionUtils;
@@ -52,7 +52,6 @@
 import org.glassfish.jersey.internal.guava.Sets;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -63,7 +62,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
-
 import static com.netease.arctic.hive.op.UpdateHiveFiles.DELETE_UNTRACKED_HIVE_FILE;
 import static com.netease.arctic.server.ArcticServiceConstants.INVALID_SNAPSHOT_ID;
 

core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java

@@ -29,14 +29,15 @@
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.ArcticFileIOs;
 import com.netease.arctic.op.ArcticHadoopTableOperations;
+import com.netease.arctic.op.CreateTableTransaction;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableBuilder;
 import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.TableMetaStore;
 import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.blocker.BasicTableBlockerManager;
 import com.netease.arctic.table.blocker.TableBlockerManager;
-import com.netease.arctic.trace.CreateTableTransaction;
 import com.netease.arctic.utils.CatalogUtil;
 import com.netease.arctic.utils.CompatiblePropertyUtil;
 import com.netease.arctic.utils.ConvertStructUtil;
@@ -76,6 +77,7 @@
   protected CatalogMeta catalogMeta;
   protected Map<String, String> customProperties;
   protected MixedTables tables;
+  protected transient TableMetaStore tableMetaStore;
 
   @Override
   public String name() {
@@ -92,6 +94,7 @@
     this.customProperties = properties;
     CatalogUtil.mergeCatalogProperties(catalogMeta, properties);
     tables = newMixedTables(catalogMeta);
+    tableMetaStore = CatalogUtil.buildMetaStore(meta);
   }
 
   protected MixedTables newMixedTables(CatalogMeta catalogMeta) {
@@ -287,18 +290,14 @@
       return table;
     }
 
-    protected ArcticTable createTableByMeta(TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
-                                            PartitionSpec partitionSpec) {
-      return tables.createTableByMeta(tableMeta, schema, primaryKeySpec, partitionSpec);
-    }
-
-    public Transaction newCreateTableTransaction() {
-      ArcticFileIO arcticFileIO = ArcticFileIOs.buildHadoopFileIO(tables.getTableMetaStore());
+    @Override
+    public Transaction createTransaction() {
+      ArcticFileIO arcticFileIO = ArcticFileIOs.buildHadoopFileIO(tableMetaStore);
       ConvertStructUtil.TableMetaBuilder builder = createTableMataBuilder();
       TableMeta meta = builder.build();
       String location = getTableLocationForCreate();
       TableOperations tableOperations = new ArcticHadoopTableOperations(new Path(location),
-          arcticFileIO, tables.getTableMetaStore().getConfiguration());
+          arcticFileIO, tableMetaStore.getConfiguration());
       TableMetadata tableMetadata = tableMetadata(schema, partitionSpec, sortOrder, properties, location);
       Transaction transaction =
           Transactions.createTableTransaction(identifier.getTableName(), tableOperations, tableMetadata);
@@ -318,6 +317,12 @@
       );
     }
 
+    protected ArcticTable createTableByMeta(
+        TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
+        PartitionSpec partitionSpec) {
+      return tables.createTableByMeta(tableMeta, schema, primaryKeySpec, partitionSpec);
+    }
+
     protected void doCreateCheck() {
       if (primaryKeySpec.primaryKeyExisted()) {
         primaryKeySpec.fieldNames().forEach(primaryKey -> {
@@ -343,17 +348,20 @@
     }
 
     protected void checkProperties() {
-      Map<String, String> mergedProperties = CatalogUtil.mergeCatalogPropertiesToTable(properties,
-          catalogMeta.getCatalogProperties());
+      Map<String, String> mergedProperties = CatalogUtil.mergeCatalogPropertiesToTable(
+          properties, catalogMeta.getCatalogProperties());
       boolean enableStream = CompatiblePropertyUtil.propertyAsBoolean(mergedProperties,
           TableProperties.ENABLE_LOG_STORE, TableProperties.ENABLE_LOG_STORE_DEFAULT);
       if (enableStream) {
-        Preconditions.checkArgument(mergedProperties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC),
+        Preconditions.checkArgument(
+            mergedProperties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC),
             "log-store.topic must not be null when log-store.enabled is true.");
-        Preconditions.checkArgument(mergedProperties.containsKey(TableProperties.LOG_STORE_ADDRESS),
+        Preconditions.checkArgument(
+            mergedProperties.containsKey(TableProperties.LOG_STORE_ADDRESS),
             "log-store.address must not be null when log-store.enabled is true.");
         String logStoreType = mergedProperties.get(LOG_STORE_TYPE);
-        Preconditions.checkArgument(logStoreType == null ||
+        Preconditions.checkArgument(
+            logStoreType == null ||
                 logStoreType.equals(LOG_STORE_STORAGE_TYPE_KAFKA) ||
                 logStoreType.equals(LOG_STORE_STORAGE_TYPE_PULSAR),
             String.format(

core/src/main/java/com/netease/arctic/catalog/IcebergCatalogWrapper.java

@@ -37,6 +37,7 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -103,12 +104,11 @@
 
     if (meta.getCatalogProperties().containsKey(CatalogMetaProperties.KEY_TABLE_FILTER)) {
       String tableFilter =
-              meta.getCatalogProperties().get(CatalogMetaProperties.KEY_TABLE_FILTER);
+          meta.getCatalogProperties().get(CatalogMetaProperties.KEY_TABLE_FILTER);
       tableFilterPattern = Pattern.compile(tableFilter);
     } else {
       tableFilterPattern = null;
     }
-
   }
 
   public IcebergCatalogWrapper(CatalogMeta meta, Map<String, String> properties) {
@@ -279,6 +279,13 @@
       return new BasicIcebergTable(identifier, table, arcticFileIO, meta.getCatalogProperties());
     }
 
+    @Override
+    public Transaction createTransaction() {
+      return icebergCatalog.newCreateTableTransaction(
+          toIcebergTableIdentifier(identifier), schema,
+          spec, properties);
+    }
+
     @Override
     public TableBuilder withPrimaryKeySpec(PrimaryKeySpec primaryKeySpec) {
       Preconditions.checkArgument(
@@ -300,7 +307,7 @@
         Table icebergTable,
         ArcticFileIO arcticFileIO,
         Map<String, String> catalogProperties) {
-      super(tableIdentifier, icebergTable, arcticFileIO, null, catalogProperties);
+      super(tableIdentifier, icebergTable, arcticFileIO, catalogProperties);
     }
 
     @Override

core/src/main/java/com/netease/arctic/catalog/MixedTables.java

@@ -1,6 +1,5 @@
 package com.netease.arctic.catalog;
 
-import com.netease.arctic.AmsClient;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
@@ -32,6 +31,9 @@
 import org.slf4j.LoggerFactory;
 import java.util.Map;
 
+/**
+ * TODO: this class will be removed when we support using restCatalog as base store for InternalCatalog
+ */
 public class MixedTables {
 
   private static final Logger LOG = LoggerFactory.getLogger(MixedTables.class);
@@ -40,19 +42,8 @@
   protected Tables tables;
   protected TableMetaStore tableMetaStore;
 
-  /**
-   * @deprecated since 0.5.0, will be removed in 0.6.0;
-   */
-  @Deprecated
-  protected AmsClient amsClient;
-
   public MixedTables(CatalogMeta catalogMeta) {
-    this(catalogMeta, null);
-  }
-
-  public MixedTables(CatalogMeta catalogMeta, AmsClient amsClient) {
     initialize(catalogMeta);
-    this.amsClient = amsClient;
   }
 
   private void initialize(CatalogMeta meta) {
@@ -94,13 +85,13 @@
     BaseTable baseTable = new BasicKeyedTable.BaseInternalTable(tableIdentifier,
         CatalogUtil.useArcticTableOperations(
             baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
-        fileIO, amsClient, catalogMeta.getCatalogProperties());
+        fileIO, catalogMeta.getCatalogProperties());
 
     Table changeIcebergTable = tableMetaStore.doAs(() -> tables.load(changeLocation));
     ChangeTable changeTable = new BasicKeyedTable.ChangeInternalTable(tableIdentifier,
         CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
             tableMetaStore.getConfiguration()),
-        fileIO, amsClient, catalogMeta.getCatalogProperties());
+        fileIO, catalogMeta.getCatalogProperties());
     PrimaryKeySpec keySpec = buildPrimaryKeySpec(baseTable.schema(), tableMeta);
     return new BasicKeyedTable(tableLocation, keySpec, baseTable, changeTable);
   }
@@ -133,22 +124,20 @@
         tableIdentifier, tableLocation, tableMeta.getProperties(),
         tableMetaStore, catalogMeta.getCatalogProperties());
     return new BasicUnkeyedTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation,
-        fileIO, tableMetaStore.getConfiguration()), fileIO, amsClient, catalogMeta.getCatalogProperties());
+        fileIO, tableMetaStore.getConfiguration()), fileIO, catalogMeta.getCatalogProperties());
   }
 
-  public ArcticTable createTableByMeta(
-      TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
-      PartitionSpec partitionSpec) {
+  public ArcticTable createTableByMeta(TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
+                                       PartitionSpec partitionSpec) {
     if (primaryKeySpec.primaryKeyExisted()) {
       return createKeyedTable(tableMeta, schema, primaryKeySpec, partitionSpec);
     } else {
       return createUnKeyedTable(tableMeta, schema, primaryKeySpec, partitionSpec);
     }
   }
 
-  protected KeyedTable createKeyedTable(
-      TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
-      PartitionSpec partitionSpec) {
+  protected KeyedTable createKeyedTable(TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
+                                        PartitionSpec partitionSpec) {
     TableIdentifier tableIdentifier = TableIdentifier.of(tableMeta.getTableIdentifier());
     String tableLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_TABLE);
     String baseLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_BASE);
@@ -168,7 +157,7 @@
     BaseTable baseTable = new BasicKeyedTable.BaseInternalTable(tableIdentifier,
         CatalogUtil.useArcticTableOperations(baseIcebergTable, baseLocation, fileIO,
             tableMetaStore.getConfiguration()),
-        fileIO, amsClient, catalogMeta.getCatalogProperties());
+        fileIO, catalogMeta.getCatalogProperties());
 
     Table changeIcebergTable = tableMetaStore.doAs(() -> {
       try {
@@ -180,7 +169,7 @@
     ChangeTable changeTable = new BasicKeyedTable.ChangeInternalTable(tableIdentifier,
         CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
             tableMetaStore.getConfiguration()),
-        fileIO, amsClient, catalogMeta.getCatalogProperties());
+        fileIO, catalogMeta.getCatalogProperties());
     return new BasicKeyedTable(tableLocation, primaryKeySpec, baseTable, changeTable);
   }
 
@@ -191,9 +180,8 @@
     tableMeta.putToProperties("flink.max-continuous-empty-commits", String.valueOf(Integer.MAX_VALUE));
   }
 
-  protected UnkeyedTable createUnKeyedTable(
-      TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
-      PartitionSpec partitionSpec) {
+  protected UnkeyedTable createUnKeyedTable(TableMeta tableMeta, Schema schema, PrimaryKeySpec primaryKeySpec,
+                                            PartitionSpec partitionSpec) {
     TableIdentifier tableIdentifier = TableIdentifier.of(tableMeta.getTableIdentifier());
     String tableLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_TABLE);
     String baseLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_BASE);
@@ -210,7 +198,7 @@
         tableIdentifier, tableLocation, tableMeta.getProperties(),
         tableMetaStore, catalogMeta.getCatalogProperties());
     return new BasicUnkeyedTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation, fileIO,
-        tableMetaStore.getConfiguration()), fileIO, amsClient, catalogMeta.getCatalogProperties());
+        tableMetaStore.getConfiguration()), fileIO, catalogMeta.getCatalogProperties());
   }
 
   public void dropTableByMeta(TableMeta tableMeta, boolean purge) {

core/src/main/java/com/netease/arctic/mixed/BasicMixedIcebergCatalog.java

@@ -24,6 +24,7 @@
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.TableTrashManagers;
+import com.netease.arctic.op.CreateTableTransaction;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableBuilder;
@@ -47,6 +48,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.thrift.TException;
+
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -304,8 +306,15 @@
     }
 
     @Override
-    public Transaction newCreateTableTransaction() {
-      return null;
+    public Transaction createTransaction() {
+      Transaction transaction = icebergCatalog.newCreateTableTransaction(
+          org.apache.iceberg.catalog.TableIdentifier.of(identifier.getDatabase(), identifier.getTableName()),
+          schema, partitionSpec, properties);
+      return new CreateTableTransaction(
+          transaction,
+          this::create,
+          () -> dropTable(identifier, true)
+      );
     }
   }
 }