[Refactor] Introduce Iceberg TableRuntimePlugin

Author: majin.nathan

amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java

@@ -54,9 +54,10 @@
 import org.apache.amoro.server.scheduler.inline.InlineTableExecutors;
 import org.apache.amoro.server.table.DefaultTableManager;
 import org.apache.amoro.server.table.DefaultTableService;
-import org.apache.amoro.server.table.RuntimeHandlerChain;
+import org.apache.amoro.server.table.IcebergTablePlugin;
 import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableRuntimeFactoryManager;
+import org.apache.amoro.server.table.TableRuntimePlugin;
 import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.server.utils.ThriftServiceProxy;
@@ -198,12 +199,11 @@ public void waitFollowerShip() throws Exception {
     haContainer.waitFollowerShip();
   }
 
-  public void startRestServices() throws Exception {
+  public void startRestServices() {
     EventsManager.getInstance();
     MetricManager.getInstance();
 
     catalogManager = new DefaultCatalogManager(serviceConfig);
-    tableManager = new DefaultTableManager(serviceConfig, catalogManager);
     optimizerManager = new DefaultOptimizerManager(serviceConfig, catalogManager);
     terminalManager = new TerminalManager(serviceConfig, catalogManager);
 
@@ -234,38 +234,37 @@ public void startOptimizingService() throws Exception {
 
     tableService =
         new DefaultTableService(serviceConfig, catalogManager, tableRuntimeFactoryManager);
-
+    tableManager = new DefaultTableManager(serviceConfig, tableService);
     optimizingService =
         new DefaultOptimizingService(serviceConfig, catalogManager, optimizerManager, tableService);
 
     processService = new ProcessService(serviceConfig, tableService);
-
-    LOG.info("Setting up AMS table executors...");
-    InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
-    addHandlerChain(optimizingService.getTableRuntimeHandler());
-    addHandlerChain(processService.getTableHandlerChain());
-    addHandlerChain(InlineTableExecutors.getInstance().getDataExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getSnapshotsExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOptimizingCommitExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOptimizingExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getBlockerExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getHiveCommitSyncExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getTableRefreshingExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getTagsAutoCreatingExecutor());
-    tableService.initialize();
+    tableService.initialize(initTablePlugins());
     LOG.info("AMS table service have been initialized");
-    tableManager.setTableService(tableService);
 
     initThriftService();
     startThriftService();
   }
 
-  private void addHandlerChain(RuntimeHandlerChain chain) {
-    if (chain != null) {
-      tableService.addHandlerChain(chain);
-    }
+  private List<TableRuntimePlugin> initTablePlugins() {
+    LOG.info("Setting up AMS table executors...");
+    InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
+    IcebergTablePlugin icebergTablePlugin =
+        IcebergTablePlugin.builder()
+            .addHandler(optimizingService.getTableRuntimeHandler())
+            .addHandler(processService.getTableHandlerChain())
+            .addHandler(InlineTableExecutors.getInstance().getDataExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getSnapshotsExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOptimizingCommitExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOptimizingExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getBlockerExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getHiveCommitSyncExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getTableRefreshingExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getTagsAutoCreatingExecutor())
+            .build();
+    return List.of(icebergTablePlugin);
   }
 
   public void disposeOptimizingService() {

amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableManager.java

@@ -26,15 +26,11 @@
 import org.apache.amoro.api.Blocker;
 import org.apache.amoro.api.TableIdentifier;
 import org.apache.amoro.config.Configurations;
-import org.apache.amoro.exception.AlreadyExistsException;
 import org.apache.amoro.exception.BlockerConflictException;
-import org.apache.amoro.exception.IllegalMetadataException;
 import org.apache.amoro.exception.ObjectNotExistsException;
 import org.apache.amoro.exception.PersistenceException;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.catalog.CatalogManager;
-import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.dashboard.model.TableOptimizingInfo;
 import org.apache.amoro.server.dashboard.utils.OptimizingUtil;
 import org.apache.amoro.server.optimizing.OptimizingTaskMeta;
@@ -54,7 +50,6 @@
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
-import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -74,18 +69,10 @@ public class DefaultTableManager extends PersistentBase implements TableManager
   public static final Logger LOG = LoggerFactory.getLogger(DefaultTableManager.class);
   private static final int TABLE_BLOCKER_RETRY = 3;
   private final long blockerTimeout;
+  private final TableService tableService;
 
-  private final CatalogManager catalogManager;
-
-  private @Nullable TableService tableService;
-
-  public DefaultTableManager(Configurations configuration, CatalogManager catalogManager) {
-    this.catalogManager = catalogManager;
+  public DefaultTableManager(Configurations configuration, TableService tableService) {
     this.blockerTimeout = configuration.get(AmoroManagementConf.BLOCKER_TIMEOUT).toMillis();
-  }
-
-  @Override
-  public void setTableService(@Nullable TableService tableService) {
     this.tableService = tableService;
   }
 
@@ -95,40 +82,12 @@ private Optional<TableService> tableService() {
 
   @Override
   public void dropTableMetadata(TableIdentifier tableIdentifier, boolean deleteData) {
-    if (StringUtils.isBlank(tableIdentifier.getTableName())) {
-      throw new IllegalMetadataException("table name is blank");
-    }
-    if (StringUtils.isBlank(tableIdentifier.getCatalog())) {
-      throw new IllegalMetadataException("catalog is blank");
-    }
-    if (StringUtils.isBlank(tableIdentifier.getDatabase())) {
-      throw new IllegalMetadataException("database is blank");
-    }
-
-    InternalCatalog internalCatalog =
-        catalogManager.getInternalCatalog(tableIdentifier.getCatalog());
-    String database = tableIdentifier.getDatabase();
-    String table = tableIdentifier.getTableName();
-    if (!internalCatalog.tableExists(database, table)) {
-      throw new ObjectNotExistsException(tableIdentifier);
-    }
-
-    ServerTableIdentifier serverTableIdentifier = internalCatalog.dropTable(database, table);
-    tableService().ifPresent(s -> s.onTableDropped(internalCatalog, serverTableIdentifier));
+    tableService().ifPresent(s -> s.dropTable(tableIdentifier));
   }
 
   @Override
   public void createTable(String catalogName, TableMetadata tableMetadata) {
-    InternalCatalog catalog = catalogManager.getInternalCatalog(catalogName);
-    String database = tableMetadata.getTableIdentifier().getDatabase();
-    String table = tableMetadata.getTableIdentifier().getTableName();
-    if (catalog.tableExists(database, table)) {
-      throw new AlreadyExistsException(
-          tableMetadata.getTableIdentifier().getIdentifier().buildTableIdentifier());
-    }
-
-    TableMetadata metadata = catalog.createTable(tableMetadata);
-    tableService().ifPresent(s -> s.onTableCreated(catalog, metadata.getTableIdentifier()));
+    tableService().ifPresent(s -> s.createTable(tableMetadata));
   }
 
   @Override

amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeFactory.java

@@ -42,7 +42,7 @@ public String name() {
   }
 
   @Override
-  public Optional<TableRuntimeCreator> accept(
+  public Optional<Creator> accept(
       ServerTableIdentifier tableIdentifier, Map<String, String> tableProperties) {
     if (tableIdentifier
         .getFormat()
@@ -52,7 +52,7 @@ public Optional<TableRuntimeCreator> accept(
     return Optional.empty();
   }
 
-  private static class TableRuntimeCreatorImpl implements TableRuntimeFactory.TableRuntimeCreator {
+  private static class TableRuntimeCreatorImpl implements TableRuntimeFactory.Creator {
     @Override
     public List<StateKey<?>> requiredStateKeys() {
       return DefaultTableRuntime.REQUIRED_STATES;