Repository: apache/amoro
Pull request #3400 (2025-01-07 13:20:27)

[improvement] encode for optimizerGroup value
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://amoro.apache.org/how-to-contribute/
  2. If the PR is related to an issue in https://github.com/apache/amoro/issues, add '[AMORO-XXXX]' in your PR title, e.g., '[AMORO-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][AMORO-XXXX] Your PR title ...'.
-->

## Why are the changes needed?

when optimizerGroup name has value like '/' , the search url will be  "http://127.0.0.1:1630/api/ams/v1/optimize/optimizerGroups/PROD/test/tables?dbSearchInput=&tableSearchInput=&page=1&pageSize=25". it will 404 for server. we should encode optimizerGroup.

when encode , it will be work well.
![image](https://github.com/user-attachments/assets/1bcb909f-70dd-47bf-934a-d4c31fc96978)


Close #xxx.

## Brief change log


-

## How was this patch tested?

- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [x] Add screenshots for manual tests if appropriate

- [ ] Run test locally before making a pull request

## Documentation

- Does this pull request introduce a new feature? (yes / no) 
 no
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)


zhoujinsong: @zhangmo8 Can you help to review this PR?
zhangmo8: Except for special characters, In the path, encoding is a default behavior of the browser. If encoding is done manually, I'm not sure whether the backend can recognize it correctly.

If the back end can correctly retrieve the character `\t`. I think this change is great.


Patch:
diff --git a/amoro-ams/pom.xml b/amoro-ams/pom.xml
index ebc862e40..b46859a22 100644
--- a/amoro-ams/pom.xml
+++ b/amoro-ams/pom.xml
@@ -200,12 +200,6 @@
             <artifactId>derby</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>mysql</groupId>
-            <artifactId>mysql-connector-java</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-dbcp2</artifactId>
@@ -403,6 +397,13 @@
             <version>${pagehelper.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <version>${mysql-connector-j.version}</version>
+            <scope>${mysql-j-dependency-scope}</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.iceberg</groupId>
             <artifactId>iceberg-data</artifactId>
@@ -451,8 +452,7 @@
             <version>1.19.6</version>
             <scope>test</scope>
         </dependency>
-
-
+        
     </dependencies>
 
     <build>
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
index c76133774..dfbb0099d 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
@@ -51,6 +51,12 @@ public class AmoroManagementConf {
           .defaultValue("admin")
           .withDescription("The administrator password");
 
+  public static final ConfigOption<Duration> CATALOG_META_CACHE_EXPIRATION_INTERVAL =
+      ConfigOptions.key("catalog-meta-cache.expiration-interval")
+          .durationType()
+          .defaultValue(Duration.ofSeconds(60))
+          .withDescription("TTL for catalog metadata.");
+
   public static final ConfigOption<Integer> TABLE_MANIFEST_IO_THREAD_COUNT =
       ConfigOptions.key("table-manifest-io.thread-count")
           .intType()
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
index 5c5ee9c46..2daa17199 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
@@ -28,6 +28,8 @@ import org.apache.amoro.api.OptimizingService;
 import org.apache.amoro.config.ConfigHelpers;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.exception.AmoroRuntimeException;
+import org.apache.amoro.server.catalog.CatalogManager;
+import org.apache.amoro.server.catalog.DefaultCatalogManager;
 import org.apache.amoro.server.dashboard.DashboardServer;
 import org.apache.amoro.server.dashboard.JavalinJsonMapper;
 import org.apache.amoro.server.dashboard.response.ErrorResponse;
@@ -92,6 +94,7 @@ public class AmoroServiceContainer {
 
   private final HighAvailabilityContainer haContainer;
   private DataSource dataSource;
+  private CatalogManager catalogManager;
   private DefaultTableService tableService;
   private DefaultOptimizingService optimizingService;
   private TerminalManager terminalManager;
@@ -146,8 +149,9 @@ public class AmoroServiceContainer {
     EventsManager.getInstance();
     MetricManager.getInstance();
 
-    tableService = new DefaultTableService(serviceConfig);
-    optimizingService = new DefaultOptimizingService(serviceConfig, tableService);
+    catalogManager = new DefaultCatalogManager(serviceConfig);
+    tableService = new DefaultTableService(serviceConfig, catalogManager);
+    optimizingService = new DefaultOptimizingService(serviceConfig, catalogManager, tableService);
 
     LOG.info("Setting up AMS table executors...");
     AsyncTableExecutors.getInstance().setup(tableService, serviceConfig);
@@ -164,7 +168,7 @@ public class AmoroServiceContainer {
     addHandlerChain(AsyncTableExecutors.getInstance().getTagsAutoCreatingExecutor());
     tableService.initialize();
     LOG.info("AMS table service have been initialized");
-    terminalManager = new TerminalManager(serviceConfig, tableService);
+    terminalManager = new TerminalManager(serviceConfig, catalogManager, tableService);
 
     initThriftService();
     startThriftService();
@@ -240,8 +244,9 @@ public class AmoroServiceContainer {
 
   private void initHttpService() {
     DashboardServer dashboardServer =
-        new DashboardServer(serviceConfig, tableService, optimizingService, terminalManager);
-    RestCatalogService restCatalogService = new RestCatalogService(tableService);
+        new DashboardServer(
+            serviceConfig, catalogManager, tableService, optimizingService, terminalManager);
+    RestCatalogService restCatalogService = new RestCatalogService(catalogManager, tableService);
 
     httpServer =
         Javalin.create(
@@ -333,7 +338,7 @@ public class AmoroServiceContainer {
         new AmoroTableMetastore.Processor<>(
             ThriftServiceProxy.createProxy(
                 AmoroTableMetastore.Iface.class,
-                new TableManagementService(tableService),
+                new TableManagementService(catalogManager, tableService),
                 AmoroRuntimeException::normalizeCompatibly));
     tableManagementServer =
         createThriftServer(
@@ -536,6 +541,11 @@ public class AmoroServiceContainer {
     return this.tableService;
   }
 
+  @VisibleForTesting
+  public CatalogManager getCatalogManager() {
+    return this.catalogManager;
+  }
+
   @VisibleForTesting
   public OptimizerManager getOptimizingService() {
     return this.optimizingService;
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
index 5a9d0c052..bc886423b 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
@@ -37,6 +37,7 @@ import org.apache.amoro.exception.TaskNotFoundException;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.resource.Resource;
 import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.optimizing.OptimizingQueue;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
 import org.apache.amoro.server.optimizing.TaskRuntime;
@@ -97,17 +98,22 @@ public class DefaultOptimizingService extends StatedPersistentBase
   private final Map<String, OptimizingQueue> optimizingQueueByToken = new ConcurrentHashMap<>();
   private final Map<String, OptimizerInstance> authOptimizers = new ConcurrentHashMap<>();
   private final OptimizerKeeper optimizerKeeper = new OptimizerKeeper();
+  private final CatalogManager catalogManager;
   private final TableService tableService;
   private final RuntimeHandlerChain tableHandlerChain;
   private final ExecutorService planExecutor;
 
-  public DefaultOptimizingService(Configurations serviceConfig, DefaultTableService tableService) {
+  public DefaultOptimizingService(
+      Configurations serviceConfig,
+      CatalogManager catalogManager,
+      DefaultTableService tableService) {
     this.optimizerTouchTimeout = serviceConfig.getLong(AmoroManagementConf.OPTIMIZER_HB_TIMEOUT);
     this.taskAckTimeout = serviceConfig.getLong(AmoroManagementConf.OPTIMIZER_TASK_ACK_TIMEOUT);
     this.maxPlanningParallelism =
         serviceConfig.getInteger(AmoroManagementConf.OPTIMIZER_MAX_PLANNING_PARALLELISM);
     this.pollingTimeout = serviceConfig.getLong(AmoroManagementConf.OPTIMIZER_POLLING_TIMEOUT);
     this.tableService = tableService;
+    this.catalogManager = catalogManager;
     this.tableHandlerChain = new TableRuntimeHandlerImpl();
     this.planExecutor =
         Executors.newCachedThreadPool(
@@ -199,7 +205,7 @@ public class DefaultOptimizingService extends StatedPersistentBase
   }
 
   private OptimizingTask extractOptimizingTask(
-      TaskRuntime task, String authToken, int threadId, OptimizingQueue queue) {
+      TaskRuntime<?> task, String authToken, int threadId, OptimizingQueue queue) {
     try {
       OptimizerThread optimizerThread = getAuthenticatedOptimizer(authToken).getThread(threadId);
       task.schedule(optimizerThread);
@@ -391,7 +397,7 @@ public class DefaultOptimizingService extends StatedPersistentBase
   }
 
   public boolean canDeleteResourceGroup(String name) {
-    for (CatalogMeta catalogMeta : tableService.listCatalogMetas()) {
+    for (CatalogMeta catalogMeta : catalogManager.listCatalogMetas()) {
       if (catalogMeta.getCatalogProperties() != null
           && catalogMeta
               .getCatalogProperties()
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java b/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
index a58bfb679..c3bd12b69 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
@@ -39,6 +39,7 @@ import org.apache.amoro.TableFormat;
 import org.apache.amoro.events.IcebergReportEvent;
 import org.apache.amoro.exception.ObjectNotExistsException;
 import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.manager.EventsManager;
@@ -106,9 +107,11 @@ public class RestCatalogService extends PersistentBase {
 
   private final JavalinJackson jsonMapper;
 
+  private final CatalogManager catalogManager;
   private final TableService tableService;
 
-  public RestCatalogService(TableService tableService) {
+  public RestCatalogService(CatalogManager catalogManager, TableService tableService) {
+    this.catalogManager = catalogManager;
     this.tableService = tableService;
     ObjectMapper objectMapper = jsonMapper();
     this.jsonMapper = new JavalinJackson(objectMapper);
@@ -432,7 +435,7 @@ public class RestCatalogService extends PersistentBase {
 
   private InternalCatalog getCatalog(String catalog) {
     Preconditions.checkNotNull(catalog, "lack required path variables: catalog");
-    ServerCatalog internalCatalog = tableService.getServerCatalog(catalog);
+    ServerCatalog internalCatalog = catalogManager.getServerCatalog(catalog);
     Preconditions.checkArgument(
         internalCatalog instanceof InternalCatalog, "The catalog is not an iceberg rest catalog");
     Set<TableFormat> tableFormats = CatalogUtil.tableFormats(internalCatalog.getMetadata());
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/TableManagementService.java b/amoro-ams/src/main/java/org/apache/amoro/server/TableManagementService.java
index 92fa90f22..4ed25538b 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/TableManagementService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/TableManagementService.java
@@ -29,6 +29,7 @@ import org.apache.amoro.api.OperationConflictException;
 import org.apache.amoro.api.TableCommitMeta;
 import org.apache.amoro.api.TableIdentifier;
 import org.apache.amoro.api.TableMeta;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.table.TableMetadata;
@@ -42,9 +43,11 @@ import java.util.stream.Collectors;
 
 public class TableManagementService implements AmoroTableMetastore.Iface {
 
+  private final CatalogManager catalogManager;
   private final TableService tableService;
 
-  public TableManagementService(TableService tableService) {
+  public TableManagementService(CatalogManager catalogManager, TableService tableService) {
+    this.catalogManager = catalogManager;
     this.tableService = tableService;
   }
 
@@ -53,29 +56,29 @@ public class TableManagementService implements AmoroTableMetastore.Iface {
 
   @Override
   public List<CatalogMeta> getCatalogs() {
-    return tableService.listCatalogMetas();
+    return catalogManager.listCatalogMetas();
   }
 
   @Override
   public CatalogMeta getCatalog(String name) {
-    return tableService.getCatalogMeta(name);
+    return catalogManager.getCatalogMeta(name);
   }
 
   @Override
   public List<String> getDatabases(String catalogName) {
-    ServerCatalog serverCatalog = tableService.getServerCatalog(catalogName);
+    ServerCatalog serverCatalog = catalogManager.getServerCatalog(catalogName);
     return serverCatalog.listDatabases();
   }
 
   @Override
   public void createDatabase(String catalogName, String database) {
-    InternalCatalog serverCatalog = tableService.getInternalCatalog(catalogName);
+    InternalCatalog serverCatalog = catalogManager.getInternalCatalog(catalogName);
     serverCatalog.createDatabase(database);
   }
 
   @Override
   public void dropDatabase(String catalogName, String database) {
-    InternalCatalog serverCatalog = tableService.getInternalCatalog(catalogName);
+    InternalCatalog serverCatalog = catalogManager.getInternalCatalog(catalogName);
     serverCatalog.dropDatabase(database);
   }
 
@@ -87,7 +90,7 @@ public class TableManagementService implements AmoroTableMetastore.Iface {
     ServerTableIdentifier identifier =
         ServerTableIdentifier.of(
             tableMeta.getTableIdentifier(), TableFormat.valueOf(tableMeta.getFormat()));
-    InternalCatalog catalog = tableService.getInternalCatalog(identifier.getCatalog());
+    InternalCatalog catalog = catalogManager.getInternalCatalog(identifier.getCatalog());
     CatalogMeta catalogMeta = catalog.getMetadata();
     TableMetadata tableMetadata = new TableMetadata(identifier, tableMeta, catalogMeta);
     tableService.createTable(catalog.name(), tableMetadata);
@@ -95,7 +98,7 @@ public class TableManagementService implements AmoroTableMetastore.Iface {
 
   @Override
   public List<TableMeta> listTables(String catalogName, String database) {
-    InternalCatalog serverCatalog = tableService.getInternalCatalog(catalogName);
+    InternalCatalog serverCatalog = catalogManager.getInternalCatalog(catalogName);
     List<TableMetadata> tableMetadataList = serverCatalog.listTableMetadataInDatabase(database);
     return tableMetadataList.stream()
         .map(TableMetadata::buildTableMeta)
@@ -104,7 +107,7 @@ public class TableManagementService implements AmoroTableMetastore.Iface {
 
   @Override
   public TableMeta getTable(TableIdentifier tableIdentifier) {
-    InternalCatalog serverCatalog = tableService.getInternalCatalog(tableIdentifier.getCatalog());
+    InternalCatalog serverCatalog = catalogManager.getInternalCatalog(tableIdentifier.getCatalog());
     TableMetadata tableMetadata =
         serverCatalog.loadTableMetadata(
             tableIdentifier.getDatabase(), tableIdentifier.getTableName());
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogService.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
similarity index 92%
rename from amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogService.java
rename to amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
index edca66f9a..35ba55bae 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
@@ -22,8 +22,8 @@ import org.apache.amoro.api.CatalogMeta;
 
 import java.util.List;
 
-/** The CatalogService interface defines the operations that can be performed on catalogs. */
-public interface CatalogService {
+/** The CatalogManager interface defines the operations that can be performed on catalogs. */
+public interface CatalogManager {
   /**
    * Returns a list of CatalogMeta objects.
    *
@@ -62,6 +62,13 @@ public interface CatalogService {
    */
   InternalCatalog getInternalCatalog(String catalogName);
 
+  /**
+   * Retrieves all ExternalCatalogs.
+   *
+   * @return a list of ExternalCatalogs
+   */
+  List<ExternalCatalog> getExternalCatalogs();
+
   /**
    * Creates a catalog based on the provided catalog meta information. The catalog name is obtained
    * from the catalog meta.
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java
new file mode 100644
index 000000000..c2ead9592
--- /dev/null
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java
@@ -0,0 +1,218 @@
+/*
+ * 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 org.apache.amoro.server.catalog;
+
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.config.Configurations;
+import org.apache.amoro.exception.AlreadyExistsException;
+import org.apache.amoro.exception.IllegalMetadataException;
+import org.apache.amoro.exception.ObjectNotExistsException;
+import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.AmoroManagementConf;
+import org.apache.amoro.server.persistence.PersistentBase;
+import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
+import org.apache.amoro.shade.guava32.com.google.common.cache.CacheBuilder;
+import org.apache.amoro.shade.guava32.com.google.common.cache.CacheLoader;
+import org.apache.amoro.shade.guava32.com.google.common.cache.LoadingCache;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class DefaultCatalogManager extends PersistentBase implements CatalogManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(DefaultCatalogManager.class);
+  protected final Configurations serverConfiguration;
+  private final LoadingCache<String, Optional<CatalogMeta>> metaCache;
+
+  private final Map<String, ServerCatalog> serverCatalogMap = Maps.newConcurrentMap();
+
+  public DefaultCatalogManager(Configurations serverConfiguration) {
+    this.serverConfiguration = serverConfiguration;
+    Duration cacheTtl =
+        serverConfiguration.get(AmoroManagementConf.CATALOG_META_CACHE_EXPIRATION_INTERVAL);
+    metaCache =
+        CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .expireAfterWrite(cacheTtl)
+            .build(
+                new CacheLoader<String, Optional<CatalogMeta>>() {
+                  @Override
+                  public @NotNull Optional<CatalogMeta> load(@NotNull String key) throws Exception {
+                    return Optional.ofNullable(
+                        getAs(CatalogMetaMapper.class, mapper -> mapper.getCatalog(key)));
+                  }
+                });
+
+    listCatalogMetas()
+        .forEach(
+            c -> {
+              ServerCatalog serverCatalog =
+                  CatalogBuilder.buildServerCatalog(c, serverConfiguration);
+              serverCatalogMap.put(c.getCatalogName(), serverCatalog);
+              metaCache.put(c.getCatalogName(), Optional.of(c));
+              LOG.info("Load catalog {}, type:{}", c.getCatalogName(), c.getCatalogType());
+            });
+    LOG.info("DefaultCatalogManager initialized, total catalogs: {}", serverCatalogMap.size());
+  }
+
+  @Override
+  public List<CatalogMeta> listCatalogMetas() {
+    return getAs(CatalogMetaMapper.class, CatalogMetaMapper::getCatalogs).stream()
+        .peek(c -> metaCache.put(c.getCatalogName(), Optional.of(c)))
+        .collect(Collectors.toList());
+  }
+
+  @Override
+  public CatalogMeta getCatalogMeta(String catalogName) {
+    return getCatalogMetaOptional(catalogName)
+        .orElseThrow(() -> new ObjectNotExistsException("Catalog " + catalogName));
+  }
+
+  private Optional<CatalogMeta> getCatalogMetaOptional(String catalogName) {
+    try {
+      return metaCache.get(catalogName);
+    } catch (ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public boolean catalogExist(String catalogName) {
+    return getCatalogMetaOptional(catalogName).isPresent();
+  }
+
+  @Override
+  public ServerCatalog getServerCatalog(String catalogName) {
+    Optional<CatalogMeta> catalogMeta = getCatalogMetaOptional(catalogName);
+    if (!catalogMeta.isPresent()) {
+      // remove if catalog is deleted
+      disposeCatalog(catalogName);
+      throw new ObjectNotExistsException("Catalog " + catalogName);
+    }
+    ServerCatalog serverCatalog =
+        serverCatalogMap.computeIfAbsent(
+            catalogName,
+            n -> CatalogBuilder.buildServerCatalog(catalogMeta.get(), serverConfiguration));
+    serverCatalog.reload(catalogMeta.get());
+    return serverCatalog;
+  }
+
+  @Override
+  public InternalCatalog getInternalCatalog(String catalogName) {
+    ServerCatalog serverCatalog = getServerCatalog(catalogName);
+    if (serverCatalog == null) {
+      throw new ObjectNotExistsException("Catalog " + catalogName);
+    }
+    if (serverCatalog.isInternal()) {
+      return (InternalCatalog) serverCatalog;
+    }
+    throw new ObjectNotExistsException("Catalog " + catalogName + " is not internal catalog");
+  }
+
+  @Override
+  public List<ExternalCatalog> getExternalCatalogs() {
+    return listCatalogMetas().stream()
+        .filter(c -> !isInternal(c))
+        .map(
+            c -> {
+              ServerCatalog serverCatalog =
+                  serverCatalogMap.computeIfAbsent(
+                      c.getCatalogName(),
+                      n -> CatalogBuilder.buildServerCatalog(c, serverConfiguration));
+              serverCatalog.reload(c);
+              return serverCatalog;
+            })
+        .map(c -> (ExternalCatalog) c)
+        .collect(Collectors.toList());
+  }
+
+  @Override
+  public void createCatalog(CatalogMeta catalogMeta) {
+    if (catalogExist(catalogMeta.getCatalogName())) {
+      throw new AlreadyExistsException("Catalog " + catalogMeta.getCatalogName());
+    }
+    // Build to make sure the catalog is valid
+    ServerCatalog catalog = CatalogBuilder.buildServerCatalog(catalogMeta, serverConfiguration);
+    doAs(CatalogMetaMapper.class, mapper -> mapper.insertCatalog(catalog.getMetadata()));
+    disposeCatalog(catalogMeta.getCatalogName());
+    serverCatalogMap.put(catalogMeta.getCatalogName(), catalog);
+    LOG.info(
+        "Create catalog {}, type:{}", catalogMeta.getCatalogName(), catalogMeta.getCatalogType());
+  }
+
+  @Override
+  public void dropCatalog(String catalogName) {
+    doAs(
+        CatalogMetaMapper.class,
+        mapper -> {
+          CatalogMeta meta = mapper.getCatalog(catalogName);
+          if (isInternal(meta)) {
+            int dbCount = mapper.selectDatabaseCount(catalogName);
+            int tblCount = mapper.selectTableCount(catalogName);
+            if (dbCount > 0 || tblCount > 0) {
+              throw new IllegalMetadataException(
+                  "Cannot drop internal catalog with databases or tables");
+            }
+          }
+          mapper.deleteCatalog(catalogName);
+          metaCache.invalidate(catalogName);
+        });
+
+    disposeCatalog(catalogName);
+  }
+
+  @Override
+  public void updateCatalog(CatalogMeta catalogMeta) {
+    ServerCatalog catalog = getServerCatalog(catalogMeta.getCatalogName());
+    validateCatalogUpdate(catalog.getMetadata(), catalogMeta);
+
+    metaCache.invalidate(catalogMeta.getCatalogName());
+    catalog.updateMetadata(catalogMeta);
+    LOG.info("Update catalog metadata: {}", catalogMeta.getCatalogName());
+  }
+
+  private void validateCatalogUpdate(CatalogMeta oldMeta, CatalogMeta newMeta) {
+    if (!oldMeta.getCatalogType().equals(newMeta.getCatalogType())) {
+      throw new IllegalMetadataException("Cannot update catalog type");
+    }
+  }
+
+  private void disposeCatalog(String name) {
+    serverCatalogMap.computeIfPresent(
+        name,
+        (n, c) -> {
+          LOG.info("Dispose catalog: {}", n);
+          c.dispose();
+          return null;
+        });
+    metaCache.invalidate(name);
+  }
+
+  private static boolean isInternal(CatalogMeta meta) {
+    return CatalogMetaProperties.CATALOG_TYPE_AMS.equalsIgnoreCase(meta.getCatalogType());
+  }
+}
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java
index 5ae37a304..8148f64a3 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java
@@ -28,8 +28,6 @@ import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
-import org.apache.amoro.table.TableMetaStore;
-import org.apache.amoro.utils.CatalogUtil;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -40,18 +38,21 @@ import java.util.stream.Collectors;
 public class ExternalCatalog extends ServerCatalog {
 
   UnifiedCatalog unifiedCatalog;
-  TableMetaStore tableMetaStore;
   private Pattern tableFilterPattern;
   private Pattern databaseFilterPattern;
 
   protected ExternalCatalog(CatalogMeta metadata) {
     super(metadata);
-    this.tableMetaStore = CatalogUtil.buildMetaStore(metadata);
     this.unifiedCatalog =
-        this.tableMetaStore.doAs(
-            () -> new CommonUnifiedCatalog(this::getMetadata, Maps.newHashMap()));
-    updateTableFilter(metadata);
-    updateDatabaseFilter(metadata);
+        metaStore.doAs(() -> new CommonUnifiedCatalog(this::getMetadata, Maps.newHashMap()));
+    catalogMetadataChanged();
+  }
+
+  @Override
+  protected void catalogMetadataChanged() {
+    this.unifiedCatalog.refresh();
+    updateDatabaseFilter(getMetadata());
+    updateTableFilter(getMetadata());
   }
 
   public void syncTable(String database, String tableName, TableFormat format) {
@@ -73,15 +74,6 @@ public class ExternalCatalog extends ServerCatalog {
         mapper -> mapper.deleteTableIdByName(getMetadata().getCatalogName(), database, tableName));
   }
 
-  @Override
-  public void updateMetadata(CatalogMeta metadata) {
-    super.updateMetadata(metadata);
-    this.tableMetaStore = CatalogUtil.buildMetaStore(metadata);
-    this.unifiedCatalog.refresh();
-    updateDatabaseFilter(metadata);
-    updateTableFilter(metadata);
-  }
-
   @Override
   public boolean databaseExists(String database) {
     return doAs(() -> unifiedCatalog.databaseExists(database));
@@ -158,6 +150,6 @@ public class ExternalCatalog extends ServerCatalog {
   }
 
   private <T> T doAs(Callable<T> callable) {
-    return tableMetaStore.doAs(callable);
+    return metaStore.doAs(callable);
   }
 }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
index 2c0a20181..b823f94b1 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
@@ -44,6 +44,11 @@ public abstract class InternalCatalog extends ServerCatalog {
     super(metadata);
   }
 
+  @Override
+  public boolean isInternal() {
+    return true;
+  }
+
   @Override
   public List<String> listDatabases() {
     return getAs(
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ServerCatalog.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ServerCatalog.java
index dfeede9c2..faa5974d8 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ServerCatalog.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ServerCatalog.java
@@ -21,18 +21,21 @@ package org.apache.amoro.server.catalog;
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableIDWithFormat;
 import org.apache.amoro.api.CatalogMeta;
-import org.apache.amoro.exception.IllegalMetadataException;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
+import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.utils.CatalogUtil;
 
 import java.util.List;
 
 public abstract class ServerCatalog extends PersistentBase {
 
   private volatile CatalogMeta metadata;
+  protected volatile TableMetaStore metaStore;
 
   protected ServerCatalog(CatalogMeta metadata) {
     this.metadata = metadata;
+    this.metaStore = CatalogUtil.buildMetaStore(metadata);
   }
 
   public String name() {
@@ -46,6 +49,26 @@ public abstract class ServerCatalog extends PersistentBase {
   public void updateMetadata(CatalogMeta metadata) {
     doAs(CatalogMetaMapper.class, mapper -> mapper.updateCatalog(metadata));
     this.metadata = metadata;
+    this.metaStore = CatalogUtil.buildMetaStore(metadata);
+    catalogMetadataChanged();
+  }
+
+  public void reload() {
+    CatalogMeta meta =
+        getAs(CatalogMetaMapper.class, mapper -> mapper.getCatalog(metadata.getCatalogName()));
+    if (meta == null) {
+      throw new IllegalStateException("Catalog " + metadata.getCatalogName() + " is dropped.");
+    }
+    this.reload(meta);
+  }
+
+  public void reload(CatalogMeta meta) {
+    if (this.metadata.equals(meta)) {
+      return;
+    }
+    this.metadata = meta;
+    this.metaStore = CatalogUtil.buildMetaStore(meta);
+    catalogMetadataChanged();
   }
 
   public abstract boolean databaseExists(String database);
@@ -61,13 +84,13 @@ public abstract class ServerCatalog extends PersistentBase {
   public abstract AmoroTable<?> loadTable(String database, String tableName);
 
   public void dispose() {
-    doAsTransaction(
-        () ->
-            doAsExisted(
-                CatalogMetaMapper.class,
-                mapper -> mapper.deleteCatalog(name()),
-                () ->
-                    new IllegalMetadataException(
-                        "Catalog " + name() + " has more than one database or table")));
+    // do resource clean up
   }
+
+  public boolean isInternal() {
+    return false;
+  }
+
+  /** Called when catalog metadata is changed. */
+  protected void catalogMetadataChanged() {}
 }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
index ccedcfaee..01c34c2c7 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
@@ -37,6 +37,7 @@ import org.apache.amoro.exception.SignatureCheckException;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.DefaultOptimizingService;
 import org.apache.amoro.server.RestCatalogService;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.dashboard.controller.CatalogController;
 import org.apache.amoro.server.dashboard.controller.HealthCheckController;
 import org.apache.amoro.server.dashboard.controller.LoginController;
@@ -96,19 +97,22 @@ public class DashboardServer {
 
   public DashboardServer(
       Configurations serviceConfig,
+      CatalogManager catalogManager,
       TableService tableService,
       DefaultOptimizingService optimizerManager,
       TerminalManager terminalManager) {
     PlatformFileManager platformFileManager = new PlatformFileManager();
-    this.catalogController = new CatalogController(tableService, platformFileManager);
+    this.catalogController = new CatalogController(catalogManager, platformFileManager);
     this.healthCheckController = new HealthCheckController();
     this.loginController = new LoginController(serviceConfig);
     this.optimizerGroupController = new OptimizerGroupController(tableService, optimizerManager);
     this.optimizerController = new OptimizerController(optimizerManager);
     this.platformFileInfoController = new PlatformFileInfoController(platformFileManager);
     this.settingController = new SettingController(serviceConfig, optimizerManager);
-    ServerTableDescriptor tableDescriptor = new ServerTableDescriptor(tableService, serviceConfig);
-    this.tableController = new TableController(tableService, tableDescriptor, serviceConfig);
+    ServerTableDescriptor tableDescriptor =
+        new ServerTableDescriptor(catalogManager, tableService, serviceConfig);
+    this.tableController =
+        new TableController(catalogManager, tableService, tableDescriptor, serviceConfig);
     this.terminalController = new TerminalController(terminalManager);
     this.versionController = new VersionController();
     this.overviewController = new OverviewController();
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/ServerTableDescriptor.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/ServerTableDescriptor.java
index 7b8ed7b8a..89e4aded9 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/ServerTableDescriptor.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/ServerTableDescriptor.java
@@ -23,6 +23,7 @@ import org.apache.amoro.TableFormat;
 import org.apache.amoro.api.TableIdentifier;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.process.ProcessStatus;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.table.TableService;
@@ -50,10 +51,13 @@ public class ServerTableDescriptor extends PersistentBase {
 
   private final Map<TableFormat, FormatTableDescriptor> formatDescriptorMap = new HashMap<>();
 
+  private final CatalogManager catalogManager;
   private final TableService tableService;
 
-  public ServerTableDescriptor(TableService tableService, Configurations serviceConfig) {
+  public ServerTableDescriptor(
+      CatalogManager catalogManager, TableService tableService, Configurations serviceConfig) {
     this.tableService = tableService;
+    this.catalogManager = catalogManager;
 
     // All table formats will jointly reuse the work thread pool named iceberg-worker-pool-%d
     ExecutorService executorService = ThreadPools.getWorkerPool();
@@ -146,7 +150,7 @@ public class ServerTableDescriptor extends PersistentBase {
   }
 
   private AmoroTable<?> loadTable(TableIdentifier identifier) {
-    ServerCatalog catalog = tableService.getServerCatalog(identifier.getCatalog());
+    ServerCatalog catalog = catalogManager.getServerCatalog(identifier.getCatalog());
     return catalog.loadTable(identifier.getDatabase(), identifier.getTableName());
   }
 }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java
index 6772d87c5..7bb783b09 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java
@@ -56,7 +56,7 @@ import org.apache.amoro.TableFormat;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.server.AmoroManagementConf;
-import org.apache.amoro.server.catalog.CatalogService;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.dashboard.PlatformFileManager;
@@ -66,7 +66,6 @@ import org.apache.amoro.server.dashboard.model.CatalogSettingInfo.ConfigFileItem
 import org.apache.amoro.server.dashboard.response.OkResponse;
 import org.apache.amoro.server.dashboard.utils.DesensitizationUtil;
 import org.apache.amoro.server.dashboard.utils.PropertiesUtil;
-import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.shade.guava32.com.google.common.base.Objects;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
@@ -166,10 +165,11 @@ public class CatalogController {
   }
 
   private final PlatformFileManager platformFileInfoService;
-  private final CatalogService tableService;
+  private final CatalogManager catalogService;
 
-  public CatalogController(TableService tableService, PlatformFileManager platformFileInfoService) {
-    this.tableService = tableService;
+  public CatalogController(
+      CatalogManager catalogService, PlatformFileManager platformFileInfoService) {
+    this.catalogService = catalogService;
     this.platformFileInfoService = platformFileInfoService;
   }
 
@@ -544,11 +544,11 @@ public class CatalogController {
   public void createCatalog(Context ctx) {
     CatalogRegisterInfo info = ctx.bodyAsClass(CatalogRegisterInfo.class);
     validateCatalogRegisterInfo(info);
-    if (tableService.catalogExist(info.getName())) {
+    if (catalogService.catalogExist(info.getName())) {
       throw new RuntimeException("Duplicate catalog name!");
     }
     CatalogMeta catalogMeta = constructCatalogMeta(info, null);
-    tableService.createCatalog(catalogMeta);
+    catalogService.createCatalog(catalogMeta);
     ctx.json(OkResponse.of(""));
   }
 
@@ -578,10 +578,10 @@ public class CatalogController {
   /** Get detail of some catalog. */
   public void getCatalogDetail(Context ctx) {
     String catalogName = ctx.pathParam("catalogName");
-    CatalogMeta catalogMeta = tableService.getCatalogMeta(catalogName);
+    CatalogMeta catalogMeta = catalogService.getCatalogMeta(catalogName);
     CatalogSettingInfo info = new CatalogSettingInfo();
 
-    if (tableService.catalogExist(catalogName)) {
+    if (catalogService.catalogExist(catalogName)) {
       info.setName(catalogMeta.getCatalogName());
       // We create ams catalog with type hadoop in v0.3, we should be compatible with it.
       if (CATALOG_TYPE_HADOOP.equals(catalogMeta.getCatalogType())
@@ -626,13 +626,13 @@ public class CatalogController {
   public void updateCatalog(Context ctx) {
     CatalogRegisterInfo info = ctx.bodyAsClass(CatalogRegisterInfo.class);
     validateCatalogRegisterInfo(info);
-    CatalogMeta optCatalog = tableService.getCatalogMeta(info.getName());
+    CatalogMeta optCatalog = catalogService.getCatalogMeta(info.getName());
     Preconditions.checkNotNull(optCatalog, "Catalog not exist!");
 
     unMaskSensitiveData(info, optCatalog);
     // check only some item can be modified!
     CatalogMeta catalogMeta = constructCatalogMeta(info, optCatalog);
-    tableService.updateCatalog(catalogMeta);
+    catalogService.updateCatalog(catalogMeta);
     ctx.json(OkResponse.ok());
   }
 
@@ -641,7 +641,7 @@ public class CatalogController {
     String catalogName = ctx.pathParam("catalogName");
     Preconditions.checkArgument(
         StringUtils.isNotEmpty(ctx.pathParam("catalogName")), "Catalog name is empty!");
-    ServerCatalog serverCatalog = tableService.getServerCatalog(catalogName);
+    ServerCatalog serverCatalog = catalogService.getServerCatalog(catalogName);
     if (serverCatalog instanceof InternalCatalog) {
       ctx.json(OkResponse.of(serverCatalog.listTables().size() == 0));
     } else {
@@ -654,7 +654,7 @@ public class CatalogController {
     String catalogName = ctx.pathParam("catalogName");
     Preconditions.checkArgument(
         StringUtils.isNotEmpty(ctx.pathParam("catalogName")), "Catalog name is empty!");
-    tableService.dropCatalog(catalogName);
+    catalogService.dropCatalog(catalogName);
     ctx.json(OkResponse.of("OK"));
   }
 
@@ -675,7 +675,7 @@ public class CatalogController {
             && StringUtils.isNotEmpty(configKey),
         "Catalog name or auth type or config key is null!");
 
-    CatalogMeta catalogMeta = tableService.getCatalogMeta(catalogName);
+    CatalogMeta catalogMeta = catalogService.getCatalogMeta(catalogName);
     if (CONFIG_TYPE_STORAGE.equalsIgnoreCase(confType)) {
       Map<String, String> storageConfig = catalogMeta.getStorageConfigs();
       String key = configKey.replaceAll("-", "\\.");
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
index 7dfb7f7c4..0606eb49f 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
@@ -36,6 +36,7 @@ import org.apache.amoro.optimizing.plan.AbstractOptimizingEvaluator;
 import org.apache.amoro.process.ProcessStatus;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.properties.HiveTableProperties;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.dashboard.ServerTableDescriptor;
 import org.apache.amoro.server.dashboard.ServerTableProperties;
@@ -99,6 +100,7 @@ public class TableController {
   private static final Logger LOG = LoggerFactory.getLogger(TableController.class);
   private static final long UPGRADE_INFO_EXPIRE_INTERVAL = 60 * 60 * 1000;
 
+  private final CatalogManager catalogManager;
   private final TableService tableService;
   private final ServerTableDescriptor tableDescriptor;
   private final Configurations serviceConfig;
@@ -107,9 +109,11 @@ public class TableController {
   private final ScheduledExecutorService tableUpgradeExecutor;
 
   public TableController(
+      CatalogManager catalogManager,
       TableService tableService,
       ServerTableDescriptor tableDescriptor,
       Configurations serviceConfig) {
+    this.catalogManager = catalogManager;
     this.tableService = tableService;
     this.tableDescriptor = tableDescriptor;
     this.serviceConfig = serviceConfig;
@@ -138,7 +142,7 @@ public class TableController {
             && StringUtils.isNotBlank(database)
             && StringUtils.isNotBlank(tableName),
         "catalog.database.tableName can not be empty in any element");
-    Preconditions.checkState(tableService.catalogExist(catalog), "invalid catalog!");
+    Preconditions.checkState(catalogManager.catalogExist(catalog), "invalid catalog!");
 
     ServerTableMeta serverTableMeta =
         tableDescriptor.getTableDetail(
@@ -178,7 +182,7 @@ public class TableController {
             && StringUtils.isNotBlank(db)
             && StringUtils.isNotBlank(table),
         "catalog.database.tableName can not be empty in any element");
-    ServerCatalog serverCatalog = tableService.getServerCatalog(catalog);
+    ServerCatalog serverCatalog = catalogManager.getServerCatalog(catalog);
     CatalogMeta catalogMeta = serverCatalog.getMetadata();
     TableMetaStore tableMetaStore = CatalogUtil.buildMetaStore(catalogMeta);
     HMSClientPool hmsClientPool =
@@ -217,7 +221,7 @@ public class TableController {
         "catalog.database.tableName can not be empty in any element");
     UpgradeHiveMeta upgradeHiveMeta = ctx.bodyAsClass(UpgradeHiveMeta.class);
 
-    ServerCatalog serverCatalog = tableService.getServerCatalog(catalog);
+    ServerCatalog serverCatalog = catalogManager.getServerCatalog(catalog);
     CatalogMeta catalogMeta = serverCatalog.getMetadata();
     String amsUri = AmsUtil.getAMSThriftAddress(serviceConfig, Constants.THRIFT_TABLE_SERVICE_NAME);
     catalogMeta.putToCatalogProperties(CatalogMetaProperties.AMS_URI, amsUri);
@@ -518,7 +522,7 @@ public class TableController {
         StringUtils.isNotBlank(catalog) && StringUtils.isNotBlank(db),
         "catalog.database can not be empty in any element");
 
-    ServerCatalog serverCatalog = tableService.getServerCatalog(catalog);
+    ServerCatalog serverCatalog = catalogManager.getServerCatalog(catalog);
     Function<TableFormat, String> formatToType =
         format -> {
           if (format.equals(TableFormat.MIXED_HIVE) || format.equals(TableFormat.MIXED_ICEBERG)) {
@@ -584,7 +588,7 @@ public class TableController {
     String keywords = ctx.queryParam("keywords");
 
     List<String> dbList =
-        tableService.getServerCatalog(catalog).listDatabases().stream()
+        catalogManager.getServerCatalog(catalog).listDatabases().stream()
             .filter(item -> StringUtils.isBlank(keywords) || item.contains(keywords))
             .collect(Collectors.toList());
     ctx.json(OkResponse.of(dbList));
@@ -596,7 +600,7 @@ public class TableController {
    * @param ctx - context for handling the request and response
    */
   public void getCatalogs(Context ctx) {
-    List<CatalogMeta> catalogs = tableService.listCatalogMetas();
+    List<CatalogMeta> catalogs = catalogManager.listCatalogMetas();
     ctx.json(OkResponse.of(catalogs));
   }
 
@@ -672,7 +676,7 @@ public class TableController {
             && StringUtils.isNotBlank(db)
             && StringUtils.isNotBlank(table),
         "catalog.database.tableName can not be empty in any element");
-    Preconditions.checkState(tableService.catalogExist(catalog), "invalid catalog!");
+    Preconditions.checkState(catalogManager.catalogExist(catalog), "invalid catalog!");
 
     ServerTableIdentifier serverTableIdentifier =
         tableService.getServerTableIdentifier(
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index 64267b8c5..2ef3b550f 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -79,7 +79,7 @@ public class OptimizingQueue extends PersistentBase {
 
   private final QuotaProvider quotaProvider;
   private final Queue<TableOptimizingProcess> tableQueue = new LinkedTransferQueue<>();
-  private final Queue<TaskRuntime<RewriteStageTask>> retryTaskQueue = new LinkedTransferQueue<>();
+  private final Queue<TaskRuntime<?>> retryTaskQueue = new LinkedTransferQueue<>();
   private final SchedulingPolicy scheduler;
   private final TableManager tableManager;
   private final Executor planExecutor;
@@ -151,6 +151,14 @@ public class OptimizingQueue extends PersistentBase {
 
   public void releaseTable(TableRuntime tableRuntime) {
     scheduler.removeTable(tableRuntime);
+    List<OptimizingProcess> processList =
+        tableQueue.stream()
+            .filter(process -> process.tableRuntime == tableRuntime)
+            .collect(Collectors.toList());
+    for (OptimizingProcess process : processList) {
+      process.close();
+      clearProcess(process);
+    }
     LOG.info(
         "Release queue {} with table {}",
         optimizerGroup.getName(),
@@ -161,15 +169,15 @@ public class OptimizingQueue extends PersistentBase {
     return scheduler.getTableRuntime(identifier) != null;
   }
 
-  private void clearProcess(TableOptimizingProcess optimizingProcess) {
+  private void clearProcess(OptimizingProcess optimizingProcess) {
     tableQueue.removeIf(process -> process.getProcessId() == optimizingProcess.getProcessId());
     retryTaskQueue.removeIf(
         taskRuntime -> taskRuntime.getTaskId().getProcessId() == optimizingProcess.getProcessId());
   }
 
-  public TaskRuntime pollTask(long maxWaitTime) {
+  public TaskRuntime<?> pollTask(long maxWaitTime) {
     long deadline = calculateDeadline(maxWaitTime);
-    TaskRuntime task = fetchTask();
+    TaskRuntime<?> task = fetchTask();
     while (task == null && waitTask(deadline)) {
       task = fetchTask();
     }
@@ -196,11 +204,12 @@ public class OptimizingQueue extends PersistentBase {
     }
   }
 
-  private TaskRuntime fetchTask() {
-    return Optional.ofNullable(retryTaskQueue.poll()).orElseGet(this::fetchScheduledTask);
+  private TaskRuntime<?> fetchTask() {
+    TaskRuntime<?> task = retryTaskQueue.poll();
+    return task != null ? task : fetchScheduledTask();
   }
 
-  private TaskRuntime fetchScheduledTask() {
+  private TaskRuntime<?> fetchScheduledTask() {
     return tableQueue.stream()
         .map(TableOptimizingProcess::poll)
         .filter(Objects::nonNull)
@@ -307,7 +316,7 @@ public class OptimizingQueue extends PersistentBase {
         .collect(Collectors.toList());
   }
 
-  public void retryTask(TaskRuntime taskRuntime) {
+  public void retryTask(TaskRuntime<?> taskRuntime) {
     taskRuntime.reset();
     retryTaskQueue.offer(taskRuntime);
   }
@@ -349,7 +358,9 @@ public class OptimizingQueue extends PersistentBase {
     return scheduler;
   }
 
-  private class TableOptimizingProcess implements OptimizingProcess, TaskRuntime.TaskOwner {
+  private class TableOptimizingProcess implements OptimizingProcess {
+
+    private final Lock lock = new ReentrantLock();
     private final long processId;
     private final OptimizingType optimizingType;
     private final TableRuntime tableRuntime;
@@ -358,7 +369,6 @@ public class OptimizingQueue extends PersistentBase {
     private final long targetChangeSnapshotId;
     private final Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> taskMap = Maps.newHashMap();
     private final Queue<TaskRuntime<RewriteStageTask>> taskQueue = new LinkedList<>();
-    private final Lock lock = new ReentrantLock();
     private volatile ProcessStatus status = ProcessStatus.RUNNING;
     private volatile String failedReason;
     private long endTime = AmoroServiceConstants.INVALID_TIME;
@@ -366,10 +376,12 @@ public class OptimizingQueue extends PersistentBase {
     private Map<String, Long> toSequence = Maps.newHashMap();
     private boolean hasCommitted = false;
 
-    public TaskRuntime poll() {
+    public TaskRuntime<?> poll() {
       lock.lock();
       try {
-        return taskQueue.poll();
+        return status != ProcessStatus.CLOSED && status != ProcessStatus.FAILED
+            ? taskQueue.poll()
+            : null;
       } finally {
         lock.unlock();
       }
@@ -431,16 +443,13 @@ public class OptimizingQueue extends PersistentBase {
         }
         this.status = ProcessStatus.CLOSED;
         this.endTime = System.currentTimeMillis();
-        persistProcessCompleted(false);
-        clearProcess(this);
+        persistAndSetCompleted(false);
       } finally {
         lock.unlock();
       }
-      releaseResourcesIfNecessary();
     }
 
-    @Override
-    public void acceptResult(TaskRuntime taskRuntime) {
+    private void acceptResult(TaskRuntime<?> taskRuntime) {
       lock.lock();
       try {
         try {
@@ -452,6 +461,10 @@ public class OptimizingQueue extends PersistentBase {
               taskRuntime.getTaskId(),
               throwable);
         }
+        // task cancel means persistAndSetCompleted has been called
+        if (taskRuntime.getStatus() == TaskRuntime.Status.CANCELED) {
+          return;
+        }
         if (isClosed()) {
           throw new OptimizingClosedException(processId);
         }
@@ -461,7 +474,6 @@ public class OptimizingQueue extends PersistentBase {
               && tableRuntime.getOptimizingStatus().isProcessing()
               && tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) {
             tableRuntime.beginCommitting();
-            clearProcess(this);
           }
         } else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) {
           if (taskRuntime.getRetry() < tableRuntime.getMaxExecuteRetryCount()) {
@@ -471,11 +483,10 @@ public class OptimizingQueue extends PersistentBase {
                 taskRuntime.getFailReason());
             retryTask(taskRuntime);
           } else {
-            clearProcess(this);
             this.failedReason = taskRuntime.getFailReason();
             this.status = ProcessStatus.FAILED;
             this.endTime = taskRuntime.getEndTime();
-            persistProcessCompleted(false);
+            persistAndSetCompleted(false);
           }
         }
       } finally {
@@ -483,14 +494,6 @@ public class OptimizingQueue extends PersistentBase {
       }
     }
 
-    // the cleanup of task should be done after unlock to avoid deadlock
-    @Override
-    public void releaseResourcesIfNecessary() {
-      if (this.status == ProcessStatus.FAILED || this.status == ProcessStatus.CLOSED) {
-        cancelTasks();
-      }
-    }
-
     @Override
     public boolean isClosed() {
       return status == ProcessStatus.CLOSED;
@@ -572,15 +575,13 @@ public class OptimizingQueue extends PersistentBase {
           buildCommit().commit();
           status = ProcessStatus.SUCCESS;
           endTime = System.currentTimeMillis();
-          persistProcessCompleted(true);
+          persistAndSetCompleted(true);
         } catch (Exception e) {
           LOG.error("{} Commit optimizing failed ", tableRuntime.getTableIdentifier(), e);
           status = ProcessStatus.FAILED;
           failedReason = ExceptionUtil.getErrorMessage(e, 4000);
           endTime = System.currentTimeMillis();
-          persistProcessCompleted(false);
-        } finally {
-          clearProcess(this);
+          persistAndSetCompleted(false);
         }
       } finally {
         lock.unlock();
@@ -654,8 +655,13 @@ public class OptimizingQueue extends PersistentBase {
           () -> tableRuntime.beginProcess(this));
     }
 
-    private void persistProcessCompleted(boolean success) {
+    private void persistAndSetCompleted(boolean success) {
       doAsTransaction(
+          () -> {
+            if (!success) {
+              cancelTasks();
+            }
+          },
           () ->
               doAs(
                   OptimizingMapper.class,
@@ -667,10 +673,10 @@ public class OptimizingQueue extends PersistentBase {
                           endTime,
                           getSummary(),
                           getFailedReason())),
-          () -> tableRuntime.completeProcess(success));
+          () -> tableRuntime.completeProcess(success),
+          () -> clearProcess(this));
     }
 
-    /** The cancellation should be invoked outside the process lock to avoid deadlock. */
     private void cancelTasks() {
       taskMap.values().forEach(TaskRuntime::tryCanceling);
     }
@@ -686,7 +692,7 @@ public class OptimizingQueue extends PersistentBase {
         Map<Integer, RewriteFilesInput> inputs = TaskFilesPersistence.loadTaskInputs(processId);
         taskRuntimes.forEach(
             taskRuntime -> {
-              taskRuntime.claimOwnership(this);
+              taskRuntime.getCompletedFuture().whenCompleted(() -> acceptResult(taskRuntime));
               taskRuntime
                   .getTaskDescriptor()
                   .setInput(inputs.get(taskRuntime.getTaskId().getTaskId()));
@@ -716,7 +722,8 @@ public class OptimizingQueue extends PersistentBase {
             tableRuntime.getTableIdentifier(),
             taskRuntime.getTaskId(),
             taskRuntime.getSummary());
-        taskMap.put(taskRuntime.getTaskId(), taskRuntime.claimOwnership(this));
+        taskRuntime.getCompletedFuture().whenCompleted(() -> acceptResult(taskRuntime));
+        taskMap.put(taskRuntime.getTaskId(), taskRuntime);
         taskQueue.offer(taskRuntime);
       }
     }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
index 85adbdcdd..59339dece 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
@@ -23,8 +23,8 @@ import org.apache.amoro.api.OptimizingTask;
 import org.apache.amoro.api.OptimizingTaskId;
 import org.apache.amoro.api.OptimizingTaskResult;
 import org.apache.amoro.exception.IllegalTaskStateException;
-import org.apache.amoro.exception.OptimizingClosedException;
 import org.apache.amoro.exception.TaskRuntimeException;
+import org.apache.amoro.process.SimpleFuture;
 import org.apache.amoro.process.StagedTaskDescriptor;
 import org.apache.amoro.server.AmoroServiceConstants;
 import org.apache.amoro.server.persistence.StatedPersistentBase;
@@ -39,10 +39,11 @@ import java.util.Set;
 
 public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends StatedPersistentBase {
 
+  private final SimpleFuture future = new SimpleFuture();
+  private final TaskStatusMachine statusMachine = new TaskStatusMachine();
   private OptimizingTaskId taskId;
   private T taskDescriptor;
   @StateField private Status status = Status.PLANNED;
-  private final TaskStatusMachine statusMachine = new TaskStatusMachine();
   @StateField private int runTimes = 0;
   @StateField private long startTime = AmoroServiceConstants.INVALID_TIME;
   @StateField private long endTime = AmoroServiceConstants.INVALID_TIME;
@@ -50,7 +51,6 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
   @StateField private String token;
   @StateField private int threadId = -1;
   @StateField private String failReason;
-  private TaskOwner owner;
 
   private TaskRuntime() {}
 
@@ -63,6 +63,10 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
     return taskDescriptor;
   }
 
+  public SimpleFuture getCompletedFuture() {
+    return future;
+  }
+
   public void complete(OptimizerThread thread, OptimizingTaskResult result) {
     invokeConsistency(
         () -> {
@@ -78,16 +82,18 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
           costTime += endTime - startTime;
           runTimes += 1;
           persistTaskRuntime();
-          owner.acceptResult(this);
+          future.complete();
           token = null;
           threadId = -1;
         });
-    owner.releaseResourcesIfNecessary();
   }
 
   void reset() {
     invokeConsistency(
         () -> {
+          if (status == Status.PLANNED) {
+            return;
+          }
           statusMachine.accept(Status.PLANNED);
           startTime = AmoroServiceConstants.INVALID_TIME;
           endTime = AmoroServiceConstants.INVALID_TIME;
@@ -95,6 +101,7 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
           threadId = -1;
           failReason = null;
           taskDescriptor.reset();
+          future.reset();
           // The cost time should not be reset since it is the total cost time of all runs.
           persistTaskRuntime();
         });
@@ -129,15 +136,11 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
               costTime += endTime - startTime;
             }
             persistTaskRuntime();
+            future.complete();
           }
         });
   }
 
-  public TaskRuntime<T> claimOwnership(TaskOwner owner) {
-    this.owner = owner;
-    return this;
-  }
-
   public boolean finished() {
     return this.status == Status.SUCCESS
         || this.status == Status.FAILED
@@ -262,25 +265,19 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
 
   private static final Map<Status, Set<Status>> nextStatusMap =
       ImmutableMap.<Status, Set<Status>>builder()
-          .put(Status.PLANNED, ImmutableSet.of(Status.PLANNED, Status.SCHEDULED, Status.CANCELED))
-          .put(
-              Status.SCHEDULED,
-              ImmutableSet.of(Status.PLANNED, Status.SCHEDULED, Status.ACKED, Status.CANCELED))
+          .put(Status.PLANNED, ImmutableSet.of(Status.SCHEDULED, Status.CANCELED))
+          .put(Status.SCHEDULED, ImmutableSet.of(Status.PLANNED, Status.ACKED, Status.CANCELED))
           .put(
               Status.ACKED,
-              ImmutableSet.of(
-                  Status.PLANNED, Status.ACKED, Status.SUCCESS, Status.FAILED, Status.CANCELED))
-          .put(Status.FAILED, ImmutableSet.of(Status.PLANNED, Status.FAILED))
-          .put(Status.SUCCESS, ImmutableSet.of(Status.SUCCESS))
-          .put(Status.CANCELED, ImmutableSet.of(Status.CANCELED))
+              ImmutableSet.of(Status.PLANNED, Status.SUCCESS, Status.FAILED, Status.CANCELED))
+          .put(Status.FAILED, ImmutableSet.of(Status.PLANNED))
+          .put(Status.CANCELED, ImmutableSet.of())
+          .put(Status.SUCCESS, ImmutableSet.of())
           .build();
 
   private class TaskStatusMachine {
 
     public void accept(Status targetStatus) {
-      if (owner.isClosed()) {
-        throw new OptimizingClosedException(taskId.getProcessId());
-      }
       if (!getNext().contains(targetStatus)) {
         throw new IllegalTaskStateException(taskId, status.name(), targetStatus.name());
       }
@@ -321,7 +318,7 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
 
     public TaskQuota() {}
 
-    public TaskQuota(TaskRuntime task) {
+    public TaskQuota(TaskRuntime<?> task) {
       this.startTime = task.getStartTime();
       this.endTime = task.getEndTime();
       this.processId = task.getTaskId().getProcessId();
@@ -367,12 +364,4 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
       return endTime <= validTime;
     }
   }
-
-  public interface TaskOwner {
-    void acceptResult(TaskRuntime taskRuntime);
-
-    void releaseResourcesIfNecessary();
-
-    boolean isClosed();
-  }
 }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
index fad12dbe4..fc551d2c6 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
@@ -27,8 +27,6 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -37,45 +35,29 @@ public abstract class StatedPersistentBase extends PersistentBase {
   private static final Map<Class<? extends PersistentBase>, Field[]> metaCache =
       Maps.newConcurrentMap();
   private static final Object NULL_VALUE = new Object();
-  private final Lock stateLock = new ReentrantLock();
   private final Field[] consistentFields;
 
   protected StatedPersistentBase() {
     consistentFields = getOrCreateConsistentFields();
   }
 
-  protected final void invokeConsistency(Runnable runnable) {
-    stateLock.lock();
+  protected void invokeConsistency(Runnable runnable) {
     Map<Field, Object> states = retainStates();
     try {
       doAsTransaction(runnable);
     } catch (Throwable throwable) {
       restoreStates(states);
       throw throwable;
-    } finally {
-      stateLock.unlock();
     }
   }
 
-  protected final <T> T invokeConsistency(Supplier<T> supplier) {
-    stateLock.lock();
+  protected <T> T invokeConsistency(Supplier<T> supplier) {
     Map<Field, Object> states = retainStates();
     try {
       return supplier.get();
     } catch (Throwable throwable) {
       restoreStates(states);
       throw throwable;
-    } finally {
-      stateLock.unlock();
-    }
-  }
-
-  protected final void invokeInStateLock(Runnable runnable) {
-    stateLock.lock();
-    try {
-      runnable.run();
-    } finally {
-      stateLock.unlock();
     }
   }
 
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/CatalogMetaMapper.java b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/CatalogMetaMapper.java
index 3d47f5197..ee62866ae 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/CatalogMetaMapper.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/CatalogMetaMapper.java
@@ -74,7 +74,7 @@ public interface CatalogMetaMapper {
         column = "catalog_properties",
         typeHandler = Map2StringConverter.class)
   })
-  List<CatalogMeta> getCatalog(@Param("catalogName") String catalogName);
+  CatalogMeta getCatalog(@Param("catalogName") String catalogName);
 
   @Insert(
       "INSERT INTO "
@@ -111,6 +111,9 @@ public interface CatalogMetaMapper {
   @Select("SELECT table_count FROM " + TABLE_NAME + " WHERE catalog_name = #{catalogName}")
   Integer selectTableCount(@Param("catalogName") String catalogName);
 
+  @Select("SELECT database_count FROM " + TABLE_NAME + " WHERE catalog_name = #{catalogName}")
+  Integer selectDatabaseCount(@Param("catalogName") String catalogName);
+
   @Update(
       "UPDATE "
           + TABLE_NAME
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
index 565cce4e0..ec65bdadc 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
@@ -22,7 +22,6 @@ import com.github.pagehelper.Page;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import org.apache.amoro.AmoroTable;
-import org.apache.amoro.NoSuchTableException;
 import org.apache.amoro.ServerTableIdentifier;
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.TableIDWithFormat;
@@ -38,15 +37,14 @@ 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.catalog.CatalogBuilder;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.ExternalCatalog;
 import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.manager.MetricManager;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
-import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.persistence.TableRuntimeMeta;
-import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
 import org.apache.amoro.server.persistence.mapper.TableBlockerMapper;
 import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
 import org.apache.amoro.server.table.blocker.TableBlocker;
@@ -84,14 +82,12 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
-public class DefaultTableService extends StatedPersistentBase implements TableService {
+public class DefaultTableService extends PersistentBase implements TableService {
 
   public static final Logger LOG = LoggerFactory.getLogger(DefaultTableService.class);
   private static final int TABLE_BLOCKER_RETRY = 3;
   private final long externalCatalogRefreshingInterval;
   private final long blockerTimeout;
-  private final Map<String, InternalCatalog> internalCatalogMap = new ConcurrentHashMap<>();
-  private final Map<String, ExternalCatalog> externalCatalogMap = new ConcurrentHashMap<>();
 
   private final Map<Long, TableRuntime> tableRuntimeMap = new ConcurrentHashMap<>();
 
@@ -103,96 +99,18 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
               .build());
   private final CompletableFuture<Boolean> initialized = new CompletableFuture<>();
   private final Configurations serverConfiguration;
+  private final CatalogManager catalogManager;
   private RuntimeHandlerChain headHandler;
   private ExecutorService tableExplorerExecutors;
 
-  public DefaultTableService(Configurations configuration) {
+  public DefaultTableService(Configurations configuration, CatalogManager catalogManager) {
+    this.catalogManager = catalogManager;
     this.externalCatalogRefreshingInterval =
         configuration.getLong(AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL);
     this.blockerTimeout = configuration.getLong(AmoroManagementConf.BLOCKER_TIMEOUT);
     this.serverConfiguration = configuration;
   }
 
-  @Override
-  public List<CatalogMeta> listCatalogMetas() {
-    checkStarted();
-    List<CatalogMeta> catalogs =
-        internalCatalogMap.values().stream()
-            .map(ServerCatalog::getMetadata)
-            .collect(Collectors.toList());
-    catalogs.addAll(
-        externalCatalogMap.values().stream()
-            .map(ServerCatalog::getMetadata)
-            .collect(Collectors.toList()));
-    return catalogs;
-  }
-
-  @Override
-  public CatalogMeta getCatalogMeta(String catalogName) {
-    checkStarted();
-    ServerCatalog catalog = getServerCatalog(catalogName);
-    return catalog.getMetadata();
-  }
-
-  @Override
-  public boolean catalogExist(String catalogName) {
-    checkStarted();
-    return internalCatalogMap.containsKey(catalogName)
-        || externalCatalogMap.containsKey(catalogName);
-  }
-
-  @Override
-  public ServerCatalog getServerCatalog(String catalogName) {
-    ServerCatalog catalog =
-        Optional.ofNullable((ServerCatalog) internalCatalogMap.get(catalogName))
-            .orElse(externalCatalogMap.get(catalogName));
-    return Optional.ofNullable(catalog)
-        .orElseThrow(() -> new ObjectNotExistsException("Catalog " + catalogName));
-  }
-
-  @Override
-  public void createCatalog(CatalogMeta catalogMeta) {
-    checkStarted();
-    if (catalogExist(catalogMeta.getCatalogName())) {
-      throw new AlreadyExistsException("Catalog " + catalogMeta.getCatalogName());
-    }
-    doAsTransaction(
-        () -> doAs(CatalogMetaMapper.class, mapper -> mapper.insertCatalog(catalogMeta)),
-        () -> initServerCatalog(catalogMeta));
-  }
-
-  private void initServerCatalog(CatalogMeta catalogMeta) {
-    ServerCatalog catalog = CatalogBuilder.buildServerCatalog(catalogMeta, serverConfiguration);
-    if (catalog instanceof InternalCatalog) {
-      internalCatalogMap.put(catalogMeta.getCatalogName(), (InternalCatalog) catalog);
-    } else {
-      externalCatalogMap.put(catalogMeta.getCatalogName(), (ExternalCatalog) catalog);
-    }
-  }
-
-  @Override
-  public void dropCatalog(String catalogName) {
-    checkStarted();
-    ServerCatalog serverCatalog = getServerCatalog(catalogName);
-    if (serverCatalog == null) {
-      throw new ObjectNotExistsException("Catalog " + catalogName);
-    }
-
-    // TableRuntime cleanup is responsibility by exploreExternalCatalog method
-    serverCatalog.dispose();
-    internalCatalogMap.remove(catalogName);
-    externalCatalogMap.remove(catalogName);
-  }
-
-  @Override
-  public void updateCatalog(CatalogMeta catalogMeta) {
-    checkStarted();
-    ServerCatalog catalog = getServerCatalog(catalogMeta.getCatalogName());
-    validateCatalogUpdate(catalog.getMetadata(), catalogMeta);
-    doAs(CatalogMetaMapper.class, mapper -> mapper.updateCatalog(catalogMeta));
-    catalog.updateMetadata(catalogMeta);
-  }
-
   @Override
   public void dropTableMetadata(TableIdentifier tableIdentifier, boolean deleteData) {
     checkStarted();
@@ -206,7 +124,8 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
       throw new IllegalMetadataException("database is blank");
     }
 
-    InternalCatalog internalCatalog = getInternalCatalog(tableIdentifier.getCatalog());
+    InternalCatalog internalCatalog =
+        catalogManager.getInternalCatalog(tableIdentifier.getCatalog());
     String database = tableIdentifier.getDatabase();
     String table = tableIdentifier.getTableName();
     if (!internalCatalog.tableExists(database, table)) {
@@ -227,7 +146,7 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
   @Override
   public void createTable(String catalogName, TableMetadata tableMetadata) {
     checkStarted();
-    InternalCatalog catalog = getInternalCatalog(catalogName);
+    InternalCatalog catalog = catalogManager.getInternalCatalog(catalogName);
     String database = tableMetadata.getTableIdentifier().getDatabase();
     String table = tableMetadata.getTableIdentifier().getTableName();
     if (catalog.tableExists(database, table)) {
@@ -248,7 +167,8 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
   @Override
   public AmoroTable<?> loadTable(ServerTableIdentifier tableIdentifier) {
     checkStarted();
-    return getServerCatalog(tableIdentifier.getCatalog())
+    return catalogManager
+        .getServerCatalog(tableIdentifier.getCatalog())
         .loadTable(tableIdentifier.getDatabase(), tableIdentifier.getTableName());
   }
 
@@ -372,11 +292,6 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
     }
   }
 
-  public InternalCatalog getInternalCatalog(String catalogName) {
-    return Optional.ofNullable(internalCatalogMap.get(catalogName))
-        .orElseThrow(() -> new ObjectNotExistsException("Catalog " + catalogName));
-  }
-
   @Override
   public void addHandlerChain(RuntimeHandlerChain handler) {
     checkNotStarted();
@@ -404,8 +319,6 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
   @Override
   public void initialize() {
     checkNotStarted();
-    List<CatalogMeta> catalogMetas = getAs(CatalogMetaMapper.class, CatalogMetaMapper::getCatalogs);
-    catalogMetas.forEach(this::initServerCatalog);
 
     List<TableRuntimeMeta> tableRuntimeMetaList =
         getAs(TableMetaMapper.class, TableMetaMapper::selectTableRuntimeMetas);
@@ -461,26 +374,6 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
             mapper.selectTableIdentifier(id.getCatalog(), id.getDatabase(), id.getTableName()));
   }
 
-  private ServerTableIdentifier getOrSyncServerTableIdentifier(TableIdentifier id) {
-    ServerTableIdentifier serverTableIdentifier = getServerTableIdentifier(id);
-    if (serverTableIdentifier != null) {
-      return serverTableIdentifier;
-    }
-    ServerCatalog serverCatalog = getServerCatalog(id.getCatalog());
-    if (serverCatalog instanceof InternalCatalog) {
-      return null;
-    }
-    try {
-      AmoroTable<?> table = serverCatalog.loadTable(id.database, id.getTableName());
-      TableIdentity identity =
-          new TableIdentity(id.getDatabase(), id.getTableName(), table.format());
-      syncTable((ExternalCatalog) serverCatalog, identity);
-      return getServerTableIdentifier(id);
-    } catch (NoSuchTableException e) {
-      return null;
-    }
-  }
-
   @Override
   public TableRuntime getRuntime(Long tableId) {
     checkStarted();
@@ -509,8 +402,11 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
       throw new IllegalStateException("TableService is not initialized");
     }
     long start = System.currentTimeMillis();
-    LOG.info("Syncing external catalogs: {}", String.join(",", externalCatalogMap.keySet()));
-    for (ExternalCatalog externalCatalog : externalCatalogMap.values()) {
+    List<ExternalCatalog> externalCatalogs = catalogManager.getExternalCatalogs();
+    List<String> externalCatalogNames =
+        externalCatalogs.stream().map(ExternalCatalog::name).collect(Collectors.toList());
+    LOG.info("Syncing external catalogs: {}", String.join(",", externalCatalogNames));
+    for (ExternalCatalog externalCatalog : externalCatalogs) {
       try {
         final List<CompletableFuture<Set<TableIdentity>>> tableIdentifiersFutures =
             Lists.newArrayList();
@@ -618,7 +514,9 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
     // It is permissible to have some erroneous states in the middle, as long as the final data is
     // consistent.
     Set<String> catalogNames =
-        listCatalogMetas().stream().map(CatalogMeta::getCatalogName).collect(Collectors.toSet());
+        catalogManager.listCatalogMetas().stream()
+            .map(CatalogMeta::getCatalogName)
+            .collect(Collectors.toSet());
     for (TableRuntime tableRuntime : tableRuntimeMap.values()) {
       if (!catalogNames.contains(tableRuntime.getTableIdentifier().getCatalog())) {
         disposeTable(tableRuntime.getTableIdentifier());
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index a21967d76..f6733af83 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -57,11 +57,13 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class TableRuntime extends StatedPersistentBase {
 
   private static final Logger LOG = LoggerFactory.getLogger(TableRuntime.class);
-
+  private final Lock tableLock = new ReentrantLock();
   private final TableRuntimeHandler tableHandler;
   private final ServerTableIdentifier tableIdentifier;
   private final List<TaskRuntime.TaskQuota> taskQuotas = new CopyOnWriteArrayList<>();
@@ -90,14 +92,10 @@ public class TableRuntime extends StatedPersistentBase {
   private final TableOptimizingMetrics optimizingMetrics;
   private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
   private final TableSummaryMetrics tableSummaryMetrics;
-
   private long targetSnapshotId;
-
   private long targetChangeSnapshotId;
-
   private Map<String, Long> fromSequence;
   private Map<String, Long> toSequence;
-
   private OptimizingType optimizingType;
 
   public TableRuntime(
@@ -176,15 +174,17 @@ public class TableRuntime extends StatedPersistentBase {
   }
 
   public void dispose() {
-    invokeInStateLock(
-        () -> {
-          doAsTransaction(
-              () -> Optional.ofNullable(optimizingProcess).ifPresent(OptimizingProcess::close),
-              () ->
-                  doAs(
-                      TableMetaMapper.class,
-                      mapper -> mapper.deleteOptimizingRuntime(tableIdentifier.getId())));
-        });
+    tableLock.lock();
+    try {
+      doAsTransaction(
+          () -> Optional.ofNullable(optimizingProcess).ifPresent(OptimizingProcess::close),
+          () ->
+              doAs(
+                  TableMetaMapper.class,
+                  mapper -> mapper.deleteOptimizingRuntime(tableIdentifier.getId())));
+    } finally {
+      tableLock.unlock();
+    }
     optimizingMetrics.unregister();
     orphanFilesCleaningMetrics.unregister();
     tableSummaryMetrics.unregister();
@@ -318,15 +318,16 @@ public class TableRuntime extends StatedPersistentBase {
    * @param startTimeMills
    */
   public void resetTaskQuotas(long startTimeMills) {
-    invokeInStateLock(
-        () -> {
-          taskQuotas.clear();
-          taskQuotas.addAll(
-              getAs(
-                  OptimizingMapper.class,
-                  mapper ->
-                      mapper.selectTaskQuotasByTime(tableIdentifier.getId(), startTimeMills)));
-        });
+    tableLock.lock();
+    try {
+      taskQuotas.clear();
+      taskQuotas.addAll(
+          getAs(
+              OptimizingMapper.class,
+              mapper -> mapper.selectTaskQuotasByTime(tableIdentifier.getId(), startTimeMills)));
+    } finally {
+      tableLock.unlock();
+    }
   }
 
   public void completeProcess(boolean success) {
@@ -646,4 +647,14 @@ public class TableRuntime extends StatedPersistentBase {
                     System.currentTimeMillis()));
     return TableBlocker.conflict(operation, tableBlockers);
   }
+
+  @Override
+  protected void invokeConsistency(Runnable runnable) {
+    tableLock.lock();
+    try {
+      super.invokeConsistency(runnable);
+    } finally {
+      tableLock.unlock();
+    }
+  }
 }
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
index e069f235f..ff359aae1 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
@@ -22,7 +22,6 @@ import org.apache.amoro.ServerTableIdentifier;
 import org.apache.amoro.api.BlockableOperation;
 import org.apache.amoro.api.Blocker;
 import org.apache.amoro.api.TableIdentifier;
-import org.apache.amoro.server.catalog.CatalogService;
 import org.apache.amoro.server.persistence.TableRuntimeMeta;
 import org.apache.commons.lang3.tuple.Pair;
 
@@ -31,7 +30,7 @@ import javax.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
 
-public interface TableService extends CatalogService, TableManager {
+public interface TableService extends TableManager {
 
   void initialize();
 
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/terminal/TerminalManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/terminal/TerminalManager.java
index 64dba7a82..ee4d95315 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/terminal/TerminalManager.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/terminal/TerminalManager.java
@@ -25,6 +25,7 @@ import org.apache.amoro.config.ConfigOptions;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.server.AmoroManagementConf;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.CatalogType;
 import org.apache.amoro.server.dashboard.model.LatestSessionInfo;
 import org.apache.amoro.server.dashboard.model.LogInfo;
@@ -59,6 +60,7 @@ public class TerminalManager {
 
   private final Configurations serviceConfig;
   private final AtomicLong threadPoolCount = new AtomicLong();
+  private final CatalogManager catalogManager;
   private final TableService tableService;
   private final TerminalSessionFactory sessionFactory;
   private final int resultLimits;
@@ -80,8 +82,10 @@ public class TerminalManager {
           new LinkedBlockingQueue<>(),
           r -> new Thread(null, r, "terminal-execute-" + threadPoolCount.incrementAndGet()));
 
-  public TerminalManager(Configurations conf, TableService tableService) {
+  public TerminalManager(
+      Configurations conf, CatalogManager catalogManager, TableService tableService) {
     this.serviceConfig = conf;
+    this.catalogManager = catalogManager;
     this.tableService = tableService;
     this.resultLimits = conf.getInteger(AmoroManagementConf.TERMINAL_RESULT_LIMIT);
     this.stopOnError = conf.getBoolean(AmoroManagementConf.TERMINAL_STOP_ON_ERROR);
@@ -101,7 +105,7 @@ public class TerminalManager {
    * @return - sessionId, session refer to a sql execution context
    */
   public String executeScript(String terminalId, String catalog, String script) {
-    CatalogMeta catalogMeta = tableService.getCatalogMeta(catalog);
+    CatalogMeta catalogMeta = catalogManager.getCatalogMeta(catalog);
     TableMetaStore metaStore = getCatalogTableMetaStore(catalogMeta);
     String sessionId = getSessionId(terminalId, metaStore, catalog);
     String connectorType = catalogConnectorType(catalogMeta);
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
index 6049621cd..f4bec4a2a 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
@@ -48,7 +48,7 @@ public class OptimizingStages {
   private static final Map<String, ProcessStage> STAGES =
       ImmutableMap.<String, ProcessStage>builder()
           .put(MINOR.getDesc(), MINOR)
-          .put(MINOR.getDesc(), MAJOR)
+          .put(MAJOR.getDesc(), MAJOR)
           .put(FULL.getDesc(), FULL)
           .put(COMMITTING.getDesc(), COMMITTING)
           .put(PLANNING.getDesc(), PLANNING)
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
index 99691cf84..c4a900dca 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
@@ -60,6 +60,7 @@ public abstract class OptimizingState extends TableProcessState {
     return watermark;
   }
 
+  @Override
   public ProcessStage getStage() {
     return stage;
   }
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
index 47be213ce..ee74627e2 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
@@ -20,6 +20,12 @@ package org.apache.amoro.process;
 
 public class ProcessStage {
 
+  /**
+   * An arbitrary stage belongs to processes which only contain one stage For those one stage
+   * processes, should transform ${@link ProcessStage} to ProcessStage for displaying
+   */
+  public static final ProcessStage ARBITRARY = new ProcessStage("", 0);
+
   /**
    * Action Stage description value, normally this value should be identical within certain actions
    */
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
index 77d2ae023..43f9471b5 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
@@ -22,9 +22,13 @@ package org.apache.amoro.process;
 public enum ProcessStatus {
   UNKNOWN,
   PENDING,
-  RUNNING,
   SUBMITTED,
+  RUNNING,
   SUCCESS,
   CLOSED,
-  FAILED
+  FAILED;
+
+  public ProcessStage toStage() {
+    return new ProcessStage(name(), ordinal());
+  }
 }
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java b/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
index ce1332bfd..f69e5e9a2 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
@@ -71,18 +71,21 @@ public class SimpleFuture {
   public void reset() {
     triggerFuture = new CompletableFuture<>();
     callbackFuture = triggerFuture;
-    whenCompleted(() -> {});
     callbackMap.keySet().forEach(this::whenCompleted);
   }
 
   /**
    * This method will trigger all callback functions in the same thread and wait until all callback
-   * functions are completed. If throws exception, completedFuture is not done.
+   * functions are completed. If throws exception, completedFuture is not done. Pay attention that
+   * reset could be called during whenCompleted runnable, should ignore waiting
    */
   public void complete() {
     try {
+      CompletableFuture<?> originalFuture = triggerFuture;
       if (triggerFuture.complete(null)) {
-        callbackFuture.get();
+        if (triggerFuture == originalFuture) {
+          callbackFuture.get();
+        }
       }
     } catch (Throwable throwable) {
       throw normalize(throwable);
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java b/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
index c58a6f4d0..335ab82b5 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
@@ -126,6 +126,10 @@ public class TableProcessState implements ProcessState {
     this.endTime = System.currentTimeMillis();
   }
 
+  public ProcessStage getStage() {
+    return status.toStage();
+  }
+
   protected void setId(long processId) {
     this.id = processId;
   }
diff --git a/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java b/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java
index 4bf9b732f..1bd0c56bb 100644
--- a/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java
+++ b/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java
@@ -25,6 +25,7 @@ import org.apache.amoro.shade.guava32.com.google.common.base.Strings;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.shade.guava32.com.google.common.hash.Hashing;
 import org.apache.amoro.shade.guava32.com.google.common.io.ByteStreams;
+import org.apache.amoro.utils.ReflectionUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -35,7 +36,6 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.util.KerberosName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import sun.security.krb5.KrbException;
 
 import java.io.ByteArrayInputStream;
 import java.io.FileOutputStream;
@@ -383,7 +383,7 @@ public class TableMetaStore implements Serializable {
             constructKerberosUgi();
           }
           LOG.info("Completed to build ugi {}", authInformation());
-        } catch (IOException | KrbException e) {
+        } catch (Exception e) {
           throw new RuntimeException("Fail to init user group information", e);
         }
       } else {
@@ -407,13 +407,17 @@ public class TableMetaStore implements Serializable {
       return ugi;
     }
 
-    private void constructKerberosUgi() throws IOException, KrbException {
+    private void constructKerberosUgi() throws Exception {
       Path confPath = generateKrbConfPath();
       String krbConfFile = saveConfInPath(confPath, KRB_CONF_FILE_NAME, krbConf);
       String keyTabFile = saveConfInPath(confPath, KEY_TAB_FILE_NAME, krbKeyTab);
       System.clearProperty(HADOOP_USER_PROPERTY);
       System.setProperty(KRB5_CONF_PROPERTY, krbConfFile);
-      sun.security.krb5.Config.refresh();
+      if (System.getProperty("java.vendor").contains("IBM")) {
+        ReflectionUtils.invoke("com.ibm.security.krb5.internal.Config", "refresh");
+      } else {
+        ReflectionUtils.invoke("sun.security.krb5.Config", "refresh");
+      }
       UserGroupInformation.setConfiguration(getConfiguration());
       KerberosName.resetDefaultRealm();
       this.ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(krbPrincipal, keyTabFile);
diff --git a/amoro-common/src/main/java/org/apache/amoro/utils/ReflectionUtils.java b/amoro-common/src/main/java/org/apache/amoro/utils/ReflectionUtils.java
new file mode 100644
index 000000000..a60e4c3f7
--- /dev/null
+++ b/amoro-common/src/main/java/org/apache/amoro/utils/ReflectionUtils.java
@@ -0,0 +1,55 @@
+/*
+ * 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 org.apache.amoro.utils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+public class ReflectionUtils {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ReflectionUtils.class);
+  private static final Map<String, Class> CLASS_MAP = new HashMap<>();
+
+  public static Object invoke(String className, String methodName)
+      throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
+    Class<?> classRef =
+        CLASS_MAP.computeIfAbsent(
+            className,
+            key -> {
+              try {
+                return Class.forName(className);
+              } catch (ClassNotFoundException e) {
+                // ignore error
+                return null;
+              }
+            });
+    if (Objects.isNull(classRef)) {
+      LOG.warn("cannot load class {}, skip execute method {}", className, methodName);
+      return null;
+    }
+    Method method = classRef.getDeclaredMethod(methodName);
+    return method.invoke(null);
+  }
+}
diff --git a/amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/main/java/org/apache/amoro/spark/iceberg/IcebergSparkFormat.java b/amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/main/java/org/apache/amoro/spark/iceberg/IcebergSparkFormat.java
index d733f9124..3bfc10bab 100644
--- a/amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/main/java/org/apache/amoro/spark/iceberg/IcebergSparkFormat.java
+++ b/amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/main/java/org/apache/amoro/spark/iceberg/IcebergSparkFormat.java
@@ -33,6 +33,7 @@ public class IcebergSparkFormat implements SparkTableFormat {
   private static final Pattern SNAPSHOT_ID = Pattern.compile("snapshot_id_(\\d+)");
   private static final Pattern BRANCH = Pattern.compile("branch_(.*)");
   private static final Pattern TAG = Pattern.compile("tag_(.*)");
+  private static final Pattern CHANGES = Pattern.compile("changes");
 
   @Override
   public TableFormat format() {
@@ -45,6 +46,7 @@ public class IcebergSparkFormat implements SparkTableFormat {
         || AT_TIMESTAMP.matcher(tableName).matches()
         || SNAPSHOT_ID.matcher(tableName).matches()
         || TAG.matcher(tableName).matches()
+        || CHANGES.matcher(tableName).matches()
         || BRANCH.matcher(tableName).matches();
   }
 
diff --git a/amoro-web/src/views/optimize/components/List.vue b/amoro-web/src/views/optimize/components/List.vue
index 487627efb..1b1fed24f 100644
--- a/amoro-web/src/views/optimize/components/List.vue
+++ b/amoro-web/src/views/optimize/components/List.vue
@@ -91,7 +91,7 @@ async function getTableList() {
   try {
     loading.value = true
     const params = {
-      optimizerGroup: optimizerGroup.value || 'all',
+      optimizerGroup: optimizerGroup.value ? encodeURIComponent(optimizerGroup.value) : 'all',
       dbSearchInput: dbSearchInput.value || '',
       tableSearchInput: tableSearchInput.value || '',
       page: pagination.current,
diff --git a/dist/src/main/amoro-bin/conf/config.yaml b/dist/src/main/amoro-bin/conf/config.yaml
index 721d990da..0e8c6f5cb 100644
--- a/dist/src/main/amoro-bin/conf/config.yaml
+++ b/dist/src/main/amoro-bin/conf/config.yaml
@@ -91,6 +91,9 @@ ams:
   table-manifest-io:
     thread-count: 20
 
+  catalog-meta-cache:
+    expiration-interval: 60s
+
   database:
     type: derby
     jdbc-driver-class: org.apache.derby.jdbc.EmbeddedDriver
diff --git a/pom.xml b/pom.xml
index 97a882b18..130552a4a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -151,10 +151,12 @@
         <guava.version>32.1.1-jre</guava.version>
         <hudi.version>0.14.1</hudi.version>
         <pagehelper.version>6.1.0</pagehelper.version>
+        <mysql-connector-j.version>9.1.0</mysql-connector-j.version>
 
         <rocksdb-dependency-scope>compile</rocksdb-dependency-scope>
         <lucene-dependency-scope>compile</lucene-dependency-scope>
         <aliyun-sdk-dependency-scope>provided</aliyun-sdk-dependency-scope>
+        <mysql-j-dependency-scope>provided</mysql-j-dependency-scope>
     </properties>
 
     <dependencies>
@@ -626,12 +628,6 @@
                 <version>${derby-jdbc.version}</version>
             </dependency>
 
-            <dependency>
-                <groupId>mysql</groupId>
-                <artifactId>mysql-connector-java</artifactId>
-                <version>${mysql-jdbc.version}</version>
-            </dependency>
-
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-dbcp2</artifactId>
@@ -1377,5 +1373,11 @@
                 <aliyun-sdk-dependency-scope>compile</aliyun-sdk-dependency-scope>
             </properties>
         </profile>
+        <profile>
+            <id>mysql-j</id>
+            <properties>
+                <mysql-j-dependency-scope>compile</mysql-j-dependency-scope>
+            </properties>
+        </profile>
     </profiles>
 </project>


Test patch:
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
index accf1c609..ed59ea58a 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
@@ -29,6 +29,7 @@ import org.apache.amoro.mixed.MixedFormatCatalog;
 import org.apache.amoro.optimizer.standalone.StandaloneOptimizer;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.server.catalog.DefaultCatalogManager;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.resource.OptimizerManager;
 import org.apache.amoro.server.resource.ResourceContainers;
@@ -67,6 +68,7 @@ public class AmsEnvironment {
   private static final String OPTIMIZE_GROUP = "default";
   private final AmoroServiceContainer serviceContainer;
   private Configurations serviceConfig;
+  private DefaultCatalogManager catalogManager;
   private DefaultTableService tableService;
   private final AtomicBoolean amsExit;
   private int tableServiceBindPort;
@@ -139,6 +141,9 @@ public class AmsEnvironment {
     startAms();
     DynFields.UnboundField<DefaultTableService> amsTableServiceField =
         DynFields.builder().hiddenImpl(AmoroServiceContainer.class, "tableService").build();
+    DynFields.UnboundField<DefaultCatalogManager> amsCatalogManagerField =
+        DynFields.builder().hiddenImpl(AmoroServiceContainer.class, "catalogManager").build();
+    catalogManager = amsCatalogManagerField.bind(serviceContainer).get();
     tableService = amsTableServiceField.bind(serviceContainer).get();
     DynFields.UnboundField<CompletableFuture<Boolean>> tableServiceField =
         DynFields.builder().hiddenImpl(DefaultTableService.class, "initialized").build();
@@ -194,7 +199,7 @@ public class AmsEnvironment {
   }
 
   public boolean tableExist(TableIdentifier tableIdentifier) {
-    ServerCatalog catalog = tableService.getServerCatalog(tableIdentifier.getCatalog());
+    ServerCatalog catalog = catalogManager.getServerCatalog(tableIdentifier.getCatalog());
     return catalog.tableExists(tableIdentifier.getDatabase(), tableIdentifier.getTableName());
   }
 
@@ -221,7 +226,7 @@ public class AmsEnvironment {
             properties,
             TableFormat.ICEBERG);
 
-    tableService.createCatalog(catalogMeta);
+    catalogManager.createCatalog(catalogMeta);
   }
 
   private void createExternalIcebergCatalog() {
@@ -235,7 +240,7 @@ public class AmsEnvironment {
             CatalogMetaProperties.CATALOG_TYPE_HADOOP,
             properties,
             TableFormat.ICEBERG);
-    tableService.createCatalog(catalogMeta);
+    catalogManager.createCatalog(catalogMeta);
   }
 
   private void createInternalMixIcebergCatalog() {
@@ -249,7 +254,7 @@ public class AmsEnvironment {
             CatalogMetaProperties.CATALOG_TYPE_AMS,
             properties,
             TableFormat.MIXED_ICEBERG);
-    tableService.createCatalog(catalogMeta);
+    catalogManager.createCatalog(catalogMeta);
     catalogs.put(
         INTERNAL_MIXED_ICEBERG_CATALOG,
         CatalogLoader.load(getTableServiceUrl() + "/" + INTERNAL_MIXED_ICEBERG_CATALOG));
@@ -260,7 +265,7 @@ public class AmsEnvironment {
     CatalogMeta catalogMeta =
         CatalogTestHelpers.buildHiveCatalogMeta(
             MIXED_HIVE_CATALOG, properties, testHMS.hiveConf(), TableFormat.MIXED_HIVE);
-    tableService.createCatalog(catalogMeta);
+    catalogManager.createCatalog(catalogMeta);
     catalogs.put(
         MIXED_HIVE_CATALOG, CatalogLoader.load(getTableServiceUrl() + "/" + MIXED_HIVE_CATALOG));
   }
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/RestCatalogServiceTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/RestCatalogServiceTestBase.java
index 99b1853fa..fe798dc78 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/RestCatalogServiceTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/RestCatalogServiceTestBase.java
@@ -23,6 +23,7 @@ import org.apache.amoro.ServerTableIdentifier;
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.InternalCatalog;
 import org.apache.amoro.server.table.TableMetadata;
 import org.apache.amoro.server.table.TableRuntime;
@@ -75,6 +76,7 @@ public abstract class RestCatalogServiceTestBase {
 
   protected abstract String catalogName();
 
+  protected CatalogManager catalogManager;
   protected TableService tableService;
   protected InternalCatalog serverCatalog;
 
@@ -82,8 +84,9 @@ public abstract class RestCatalogServiceTestBase {
 
   @BeforeEach
   public void before() {
+    catalogManager = ams.serviceContainer().getCatalogManager();
     tableService = ams.serviceContainer().getTableService();
-    serverCatalog = (InternalCatalog) tableService.getServerCatalog(catalogName());
+    serverCatalog = catalogManager.getInternalCatalog(catalogName());
     location =
         serverCatalog.getMetadata().getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE)
             + "/"
@@ -114,7 +117,7 @@ public abstract class RestCatalogServiceTestBase {
   }
 
   protected TableMetadata getTableMetadata(TableIdentifier identifier) {
-    InternalCatalog internalCatalog = tableService.getInternalCatalog(identifier.getCatalog());
+    InternalCatalog internalCatalog = catalogManager.getInternalCatalog(identifier.getCatalog());
     return internalCatalog.loadTableMetadata(identifier.getDatabase(), identifier.getTableName());
   }
 
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
index b780f6107..a5c406766 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
@@ -186,7 +186,7 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
     optimizingService().ackTask(token, THREAD_ID, task.getTaskId());
     assertTaskStatus(TaskRuntime.Status.ACKED);
 
-    TaskRuntime taskRuntime =
+    TaskRuntime<?> taskRuntime =
         optimizingService().listTasks(defaultResourceGroup().getName()).get(0);
     optimizingService().completeTask(token, buildOptimizingTaskResult(task.getTaskId()));
     assertTaskCompleted(taskRuntime);
@@ -310,7 +310,7 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
     reload();
     assertTaskStatus(TaskRuntime.Status.ACKED);
 
-    TaskRuntime taskRuntime =
+    TaskRuntime<?> taskRuntime =
         optimizingService().listTasks(defaultResourceGroup().getName()).get(0);
     optimizingService().completeTask(token, buildOptimizingTaskResult(task.getTaskId()));
     assertTaskCompleted(taskRuntime);
@@ -658,12 +658,10 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
         optimizingService().listTasks(defaultResourceGroup().getName()).get(0).getStatus());
   }
 
-  private void assertTaskCompleted(TaskRuntime taskRuntime) {
+  private void assertTaskCompleted(TaskRuntime<?> taskRuntime) {
     if (taskRuntime != null) {
       Assertions.assertEquals(TaskRuntime.Status.SUCCESS, taskRuntime.getStatus());
     }
-    Assertions.assertEquals(
-        0, optimizingService().listTasks(defaultResourceGroup().getName()).size());
     Assertions.assertEquals(
         ProcessStatus.RUNNING,
         tableService()
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
index 598254073..936b54c5b 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
@@ -75,7 +75,7 @@ public class TestInternalIcebergCatalogService extends RestCatalogServiceTestBas
       CatalogMeta oldMeta = meta.deepCopy();
       meta.putToCatalogProperties("cache-enabled", "false");
       meta.putToCatalogProperties("cache.expiration-interval-ms", "10000");
-      serverCatalog.updateMetadata(meta);
+      catalogManager.updateCatalog(meta);
       String warehouseInAMS = meta.getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE);
 
       Map<String, String> clientSideConfiguration = Maps.newHashMap();
@@ -91,7 +91,7 @@ public class TestInternalIcebergCatalogService extends RestCatalogServiceTestBas
       } catch (IOException e) {
         throw new RuntimeException(e);
       } finally {
-        serverCatalog.updateMetadata(oldMeta);
+        catalogManager.updateCatalog(oldMeta);
       }
     }
   }
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TableCatalogTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TableCatalogTestBase.java
index 3a7eb3731..a50d911f3 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TableCatalogTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TableCatalogTestBase.java
@@ -52,13 +52,13 @@ public class TableCatalogTestBase extends TableServiceTestBase {
     amoroCatalogTestHelper.initWarehouse(path);
     amoroCatalogTestHelper.initHiveConf(TEST_HMS.getHiveConf());
     this.amoroCatalog = amoroCatalogTestHelper.amoroCatalog();
-    tableService().createCatalog(amoroCatalogTestHelper.getCatalogMeta());
+    CATALOG_MANAGER.createCatalog(amoroCatalogTestHelper.getCatalogMeta());
     this.originalCatalog = amoroCatalogTestHelper.originalCatalog();
   }
 
   @After
   public void clean() {
-    tableService().dropCatalog(amoroCatalogTestHelper.catalogName());
+    CATALOG_MANAGER.dropCatalog(amoroCatalogTestHelper.catalogName());
     amoroCatalogTestHelper.clean();
   }
 
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestServerCatalog.java b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestServerCatalog.java
index 1e41c1ede..675ba6d46 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestServerCatalog.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestServerCatalog.java
@@ -103,7 +103,7 @@ public class TestServerCatalog extends TableCatalogTestBase {
     metadata
         .getCatalogProperties()
         .put(CatalogMetaProperties.KEY_TABLE_FILTER, dbWithFilter + "." + tableWithFilter1);
-    getServerCatalog().updateMetadata(metadata);
+    CATALOG_MANAGER.updateCatalog(metadata);
     Assert.assertEquals(1, getServerCatalog().listTables(dbWithFilter).size());
     Assert.assertEquals(
         tableWithFilter1,
@@ -113,20 +113,20 @@ public class TestServerCatalog extends TableCatalogTestBase {
     metadata
         .getCatalogProperties()
         .put(CatalogMetaProperties.KEY_TABLE_FILTER, dbWithFilter + "\\." + ".+");
-    getServerCatalog().updateMetadata(metadata2);
+    CATALOG_MANAGER.updateCatalog(metadata2);
     Assert.assertEquals(2, getServerCatalog().listTables(dbWithFilter).size());
 
     CatalogMeta metadata3 = getServerCatalog().getMetadata();
     metadata
         .getCatalogProperties()
         .put(CatalogMetaProperties.KEY_TABLE_FILTER, testDatabaseName + "\\." + ".+");
-    getServerCatalog().updateMetadata(metadata3);
+    CATALOG_MANAGER.updateCatalog(metadata3);
     Assert.assertEquals(1, getServerCatalog().listTables(testDatabaseName).size());
     Assert.assertTrue(getServerCatalog().listTables(dbWithFilter).isEmpty());
 
     CatalogMeta metadata4 = getServerCatalog().getMetadata();
     metadata.getCatalogProperties().remove(CatalogMetaProperties.KEY_TABLE_FILTER);
-    getServerCatalog().updateMetadata(metadata4);
+    CATALOG_MANAGER.updateCatalog(metadata4);
   }
 
   @Test
@@ -135,6 +135,6 @@ public class TestServerCatalog extends TableCatalogTestBase {
   }
 
   private ServerCatalog getServerCatalog() {
-    return tableService().getServerCatalog(getAmoroCatalogTestHelper().catalogName());
+    return CATALOG_MANAGER.getServerCatalog(getAmoroCatalogTestHelper().catalogName());
   }
 }
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergHadoopOptimizing.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergHadoopOptimizing.java
index ec8dbd01f..29ab60d53 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergHadoopOptimizing.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergHadoopOptimizing.java
@@ -407,7 +407,7 @@ public class TestIcebergHadoopOptimizing extends AbstractOptimizingTest {
 
   private Table createIcebergTable(String catalog, PartitionSpec spec, int formatVersion) {
     ServerCatalog serverCatalog =
-        amsEnv.serviceContainer().getTableService().getServerCatalog(catalog);
+        amsEnv.serviceContainer().getCatalogManager().getServerCatalog(catalog);
     if (serverCatalog instanceof InternalCatalog) {
       this.serverCatalog = (InternalCatalog) serverCatalog;
       if (!this.serverCatalog.databaseExists(DATABASE)) {
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index bf668fbcf..eeeba78c3 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -174,12 +174,12 @@ public class TestOptimizingQueue extends AMSTableTestBase {
     OptimizingQueue queue = buildOptimizingGroupService(tableRuntimeMeta);
 
     // 1.poll task
-    TaskRuntime task = queue.pollTask(MAX_POLLING_TIME);
+    TaskRuntime<?> task = queue.pollTask(MAX_POLLING_TIME);
     Assert.assertNotNull(task);
 
     for (int i = 0; i < TableProperties.SELF_OPTIMIZING_EXECUTE_RETRY_NUMBER_DEFAULT; i++) {
       queue.retryTask(task);
-      TaskRuntime retryTask = queue.pollTask(MAX_POLLING_TIME);
+      TaskRuntime<?> retryTask = queue.pollTask(MAX_POLLING_TIME);
       Assert.assertEquals(retryTask.getTaskId(), task.getTaskId());
       retryTask.schedule(optimizerThread);
       retryTask.ack(optimizerThread);
@@ -190,7 +190,7 @@ public class TestOptimizingQueue extends AMSTableTestBase {
     }
 
     queue.retryTask(task);
-    TaskRuntime retryTask = queue.pollTask(MAX_POLLING_TIME);
+    TaskRuntime<?> retryTask = queue.pollTask(MAX_POLLING_TIME);
     Assert.assertEquals(retryTask.getTaskId(), task.getTaskId());
     retryTask.schedule(optimizerThread);
     retryTask.ack(optimizerThread);
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/AMSTableTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/AMSTableTestBase.java
index df7e478cf..2b13018f1 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/AMSTableTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/AMSTableTestBase.java
@@ -96,7 +96,7 @@ public class AMSTableTestBase extends TableServiceTestBase {
         }
       }
 
-      tableService().createCatalog(catalogMeta);
+      CATALOG_MANAGER.createCatalog(catalogMeta);
       try {
         Database database = new Database();
         database.setName(TableTestHelper.TEST_DB_NAME);
@@ -120,7 +120,7 @@ public class AMSTableTestBase extends TableServiceTestBase {
       dropDatabase();
     }
     if (catalogMeta != null) {
-      tableService().dropCatalog(catalogMeta.getCatalogName());
+      CATALOG_MANAGER.dropCatalog(catalogMeta.getCatalogName());
       TEST_HMS.getHiveClient().dropDatabase(TableTestHelper.TEST_DB_NAME, false, true);
     }
   }
@@ -146,7 +146,7 @@ public class AMSTableTestBase extends TableServiceTestBase {
   protected void createDatabase() {
     if (externalCatalog == null) {
       InternalCatalog catalog =
-          tableService().getInternalCatalog(TableTestHelper.TEST_CATALOG_NAME);
+          CATALOG_MANAGER.getInternalCatalog(TableTestHelper.TEST_CATALOG_NAME);
       if (!catalog.listDatabases().contains(TableTestHelper.TEST_DB_NAME)) {
         catalog.createDatabase(TableTestHelper.TEST_DB_NAME);
       }
@@ -161,7 +161,7 @@ public class AMSTableTestBase extends TableServiceTestBase {
   protected void dropDatabase() {
     if (externalCatalog == null) {
       InternalCatalog catalog =
-          tableService().getInternalCatalog(TableTestHelper.TEST_CATALOG_NAME);
+          CATALOG_MANAGER.getInternalCatalog(TableTestHelper.TEST_CATALOG_NAME);
       catalog.dropDatabase(TableTestHelper.TEST_DB_NAME);
     } else {
       externalCatalog.dropDatabase(TableTestHelper.TEST_DB_NAME);
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TableServiceTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TableServiceTestBase.java
index 8dd33ff77..0138e4e4f 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/TableServiceTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TableServiceTestBase.java
@@ -22,6 +22,7 @@ import org.apache.amoro.config.Configurations;
 import org.apache.amoro.resource.ResourceGroup;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.DefaultOptimizingService;
+import org.apache.amoro.server.catalog.DefaultCatalogManager;
 import org.apache.amoro.server.manager.EventsManager;
 import org.apache.amoro.server.manager.MetricManager;
 import org.junit.AfterClass;
@@ -33,6 +34,7 @@ public abstract class TableServiceTestBase {
 
   @ClassRule public static DerbyPersistence DERBY = new DerbyPersistence();
 
+  protected static DefaultCatalogManager CATALOG_MANAGER = null;
   private static DefaultTableService TABLE_SERVICE = null;
   private static DefaultOptimizingService OPTIMIZING_SERVICE = null;
 
@@ -41,8 +43,10 @@ public abstract class TableServiceTestBase {
     try {
       Configurations configurations = new Configurations();
       configurations.set(AmoroManagementConf.OPTIMIZER_HB_TIMEOUT, 800L);
-      TABLE_SERVICE = new DefaultTableService(new Configurations());
-      OPTIMIZING_SERVICE = new DefaultOptimizingService(configurations, TABLE_SERVICE);
+      CATALOG_MANAGER = new DefaultCatalogManager(configurations);
+      TABLE_SERVICE = new DefaultTableService(new Configurations(), CATALOG_MANAGER);
+      OPTIMIZING_SERVICE =
+          new DefaultOptimizingService(configurations, CATALOG_MANAGER, TABLE_SERVICE);
       TABLE_SERVICE.addHandlerChain(OPTIMIZING_SERVICE.getTableRuntimeHandler());
       TABLE_SERVICE.initialize();
       try {
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestCatalogService.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestCatalogService.java
index fd6a5def6..48362f65d 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestCatalogService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestCatalogService.java
@@ -60,18 +60,18 @@ public class TestCatalogService extends TableServiceTestBase {
   public void testCreateAndDropCatalog() {
     CatalogMeta catalogMeta = catalogTestHelper.buildCatalogMeta("/tmp");
     // test create catalog
-    tableService().createCatalog(catalogMeta);
+    CATALOG_MANAGER.createCatalog(catalogMeta);
 
     // test create duplicate catalog
     Assert.assertThrows(
-        AlreadyExistsException.class, () -> tableService().createCatalog(catalogMeta));
+        AlreadyExistsException.class, () -> CATALOG_MANAGER.createCatalog(catalogMeta));
 
     // test get catalog
-    CatalogMeta readCatalogMeta = tableService().getCatalogMeta(catalogMeta.getCatalogName());
+    CatalogMeta readCatalogMeta = CATALOG_MANAGER.getCatalogMeta(catalogMeta.getCatalogName());
     Assert.assertEquals(catalogMeta, readCatalogMeta);
 
     // test get catalog list
-    List<CatalogMeta> catalogMetas = tableService().listCatalogMetas();
+    List<CatalogMeta> catalogMetas = CATALOG_MANAGER.listCatalogMetas();
     Assert.assertEquals(1, catalogMetas.size());
     Assert.assertEquals(
         catalogMeta,
@@ -81,57 +81,57 @@ public class TestCatalogService extends TableServiceTestBase {
             .orElseThrow(() -> new IllegalStateException("Cannot find expect catalog")));
 
     // test catalogExist
-    Assert.assertTrue(tableService().catalogExist(catalogMeta.getCatalogName()));
+    Assert.assertTrue(CATALOG_MANAGER.catalogExist(catalogMeta.getCatalogName()));
 
     // test drop catalog
-    tableService().dropCatalog(catalogMeta.getCatalogName());
+    CATALOG_MANAGER.dropCatalog(catalogMeta.getCatalogName());
 
     // test drop not existed catalog
     Assert.assertThrows(
         ObjectNotExistsException.class,
-        () -> tableService().getCatalogMeta(catalogMeta.getCatalogName()));
+        () -> CATALOG_MANAGER.getCatalogMeta(catalogMeta.getCatalogName()));
 
-    Assert.assertFalse(tableService().catalogExist(catalogMeta.getCatalogName()));
+    Assert.assertFalse(CATALOG_MANAGER.catalogExist(catalogMeta.getCatalogName()));
   }
 
   @Test
   public void testUpdateCatalog() {
     CatalogMeta catalogMeta = catalogTestHelper.buildCatalogMeta("/tmp");
-    tableService().createCatalog(catalogMeta);
+    CATALOG_MANAGER.createCatalog(catalogMeta);
 
     CatalogMeta updateCatalogMeta = new CatalogMeta(catalogMeta);
     updateCatalogMeta.getCatalogProperties().put("k2", "V2");
     updateCatalogMeta.getCatalogProperties().put("k3", "v3");
-    tableService().updateCatalog(updateCatalogMeta);
-    CatalogMeta getCatalogMeta = tableService().getCatalogMeta(catalogMeta.getCatalogName());
+    CATALOG_MANAGER.updateCatalog(updateCatalogMeta);
+    CatalogMeta getCatalogMeta = CATALOG_MANAGER.getCatalogMeta(catalogMeta.getCatalogName());
     Assert.assertEquals("V2", getCatalogMeta.getCatalogProperties().get("k2"));
     Assert.assertEquals("v3", getCatalogMeta.getCatalogProperties().get("k3"));
     Assert.assertEquals(
-        updateCatalogMeta, tableService().getCatalogMeta(catalogMeta.getCatalogName()));
+        updateCatalogMeta, CATALOG_MANAGER.getCatalogMeta(catalogMeta.getCatalogName()));
 
     // test update catalog type
     final CatalogMeta updateCatalogMeta2 = new CatalogMeta(updateCatalogMeta);
     updateCatalogMeta2.setCatalogType(CatalogMetaProperties.CATALOG_TYPE_CUSTOM);
     Assert.assertThrows(
-        IllegalMetadataException.class, () -> tableService().updateCatalog(updateCatalogMeta2));
+        IllegalMetadataException.class, () -> CATALOG_MANAGER.updateCatalog(updateCatalogMeta2));
 
     // test update unknown catalog
-    tableService().dropCatalog(catalogMeta.getCatalogName());
+    CATALOG_MANAGER.dropCatalog(catalogMeta.getCatalogName());
     Assert.assertThrows(
-        ObjectNotExistsException.class, () -> tableService().updateCatalog(catalogMeta));
+        ObjectNotExistsException.class, () -> CATALOG_MANAGER.updateCatalog(catalogMeta));
   }
 
   @Test
   public void testDropCatalogWithDatabase() {
     Assume.assumeTrue(catalogTestHelper.tableFormat().equals(TableFormat.MIXED_ICEBERG));
     CatalogMeta catalogMeta = catalogTestHelper.buildCatalogMeta("/tmp");
-    tableService().createCatalog(catalogMeta);
-    InternalCatalog catalog = tableService().getInternalCatalog(catalogMeta.getCatalogName());
+    CATALOG_MANAGER.createCatalog(catalogMeta);
+    InternalCatalog catalog = CATALOG_MANAGER.getInternalCatalog(catalogMeta.getCatalogName());
     catalog.createDatabase("test_db");
     Assert.assertThrows(
         IllegalMetadataException.class,
-        () -> tableService().dropCatalog(catalogMeta.getCatalogName()));
+        () -> CATALOG_MANAGER.dropCatalog(catalogMeta.getCatalogName()));
     catalog.dropDatabase("test_db");
-    tableService().dropCatalog(catalogMeta.getCatalogName());
+    CATALOG_MANAGER.dropCatalog(catalogMeta.getCatalogName());
   }
 }
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDatabaseService.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDatabaseService.java
index 885db19b4..aadac0f92 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDatabaseService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDatabaseService.java
@@ -56,7 +56,7 @@ public class TestDatabaseService extends AMSTableTestBase {
 
   @Test
   public void testCreateAndDropDatabase() {
-    InternalCatalog catalog = tableService().getInternalCatalog(TEST_CATALOG_NAME);
+    InternalCatalog catalog = CATALOG_MANAGER.getInternalCatalog(TEST_CATALOG_NAME);
     // test create database
     catalog.createDatabase(TEST_DB_NAME);
 
@@ -78,7 +78,7 @@ public class TestDatabaseService extends AMSTableTestBase {
   public void testDropDatabaseWithTable() {
     Assume.assumeTrue(catalogTestHelper().tableFormat().equals(TableFormat.MIXED_ICEBERG));
     Assume.assumeTrue(catalogTestHelper().isInternalCatalog());
-    InternalCatalog catalog = tableService().getInternalCatalog(TEST_CATALOG_NAME);
+    InternalCatalog catalog = CATALOG_MANAGER.getInternalCatalog(TEST_CATALOG_NAME);
     catalog.createDatabase(TEST_DB_NAME);
 
     createTable();
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableRuntimeHandler.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableRuntimeHandler.java
index 641ac0e86..8329d8e7f 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableRuntimeHandler.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableRuntimeHandler.java
@@ -67,7 +67,7 @@ public class TestTableRuntimeHandler extends AMSTableTestBase {
 
   @Test
   public void testInitialize() throws Exception {
-    tableService = new DefaultTableService(new Configurations());
+    tableService = new DefaultTableService(new Configurations(), CATALOG_MANAGER);
     TestHandler handler = new TestHandler();
     tableService.addHandlerChain(handler);
     tableService.initialize();
@@ -86,7 +86,7 @@ public class TestTableRuntimeHandler extends AMSTableTestBase {
     Assert.assertTrue(handler.isDisposed());
 
     // initialize with a history table
-    tableService = new DefaultTableService(new Configurations());
+    tableService = new DefaultTableService(new Configurations(), CATALOG_MANAGER);
     handler = new TestHandler();
     tableService.addHandlerChain(handler);
     tableService.initialize();
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableService.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableService.java
index a1e7a59a0..b20e8f6f2 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableService.java
@@ -74,7 +74,7 @@ public class TestTableService extends AMSTableTestBase {
 
   @Test
   public void testCreateAndDropTable() {
-    ServerCatalog serverCatalog = tableService().getServerCatalog(TEST_CATALOG_NAME);
+    ServerCatalog serverCatalog = CATALOG_MANAGER.getServerCatalog(TEST_CATALOG_NAME);
     InternalCatalog internalCatalog =
         catalogTestHelper().isInternalCatalog() ? (InternalCatalog) serverCatalog : null;
 
@@ -94,7 +94,7 @@ public class TestTableService extends AMSTableTestBase {
 
     // test list tables
     List<TableIDWithFormat> tableIdentifierList =
-        tableService().getServerCatalog(TEST_CATALOG_NAME).listTables(TEST_DB_NAME);
+        CATALOG_MANAGER.getServerCatalog(TEST_CATALOG_NAME).listTables(TEST_DB_NAME);
     Assert.assertEquals(1, tableIdentifierList.size());
     Assert.assertEquals(
         tableMeta().getTableIdentifier(),
diff --git a/amoro-common/src/test/java/org/apache/amoro/utils/ReflectionUtilsTest.java b/amoro-common/src/test/java/org/apache/amoro/utils/ReflectionUtilsTest.java
new file mode 100644
index 000000000..46931101d
--- /dev/null
+++ b/amoro-common/src/test/java/org/apache/amoro/utils/ReflectionUtilsTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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 org.apache.amoro.utils;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import org.junit.jupiter.api.Test;
+
+class ReflectionUtilsTest {
+
+  @Test
+  void testInvokeValidClassAndMethod() {
+    try {
+      // Invoke the static method in TestClass
+      Object result = ReflectionUtils.invoke("java.lang.System", "currentTimeMillis");
+      assertNotNull(result);
+    } catch (Exception e) {
+      fail("Invocation failed: " + e.getMessage());
+    }
+  }
+
+  @Test
+  void testInvokeNonExistentClass() {
+    try {
+      Object result =
+          ReflectionUtils.invoke("com.example.nonexistent.NonExistentClass", "testMethod");
+      assertNull(result); // The method should return null for non-existent class
+    } catch (Exception e) {
+      fail("Exception should not be thrown for non-existent class: " + e.getMessage());
+    }
+  }
+}
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
index 363b1042e..60c5fba77 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
@@ -95,7 +95,7 @@ public class RewriteStageTask
   }
 
   private static long getRecordCnt(ContentFile<?>[] contentFiles) {
-    int recordCnt = 0;
+    long recordCnt = 0;
     if (contentFiles != null) {
       for (ContentFile<?> contentFile : contentFiles) {
         recordCnt += contentFile.recordCount();