complete

apache · Jan 14, 2025 · 3f4c8de · 3f4c8de
1 parent f0e752e
commit 3f4c8de
Show file tree

Hide file tree

Showing 47 changed files with 367 additions and 874 deletions.
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
@@ -45,11 +45,9 @@
 import org.apache.amoro.server.resource.ResourceContainers;
 import org.apache.amoro.server.table.DefaultTableManager;
 import org.apache.amoro.server.table.DefaultTableService;
-import org.apache.amoro.server.table.DefaultTableServiceOld;
 import org.apache.amoro.server.table.RuntimeHandlerChain;
 import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableService;
-import org.apache.amoro.server.table.TableServiceOld;
 import org.apache.amoro.server.table.executor.AsyncTableExecutors;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.server.utils.ThriftServiceProxy;
@@ -155,10 +153,11 @@ public void startService() throws Exception {
     MetricManager.getInstance();
 
     catalogManager = new DefaultCatalogManager(serviceConfig);
-
     tableManager = new DefaultTableManager(serviceConfig, catalogManager);
+
     tableService = new DefaultTableService(serviceConfig, catalogManager);
-    optimizingService = new DefaultOptimizingService(serviceConfig, catalogManager, tableService);
+    optimizingService =
+        new DefaultOptimizingService(serviceConfig, catalogManager, tableManager, tableService);
 
     LOG.info("Setting up AMS table executors...");
     AsyncTableExecutors.getInstance().setup(tableService, serviceConfig);
@@ -253,7 +252,12 @@ private void startThriftServer(TServer server, String threadName) {
   private void initHttpService() {
     DashboardServer dashboardServer =
         new DashboardServer(
-            serviceConfig, catalogManager, tableManager, optimizingService, terminalManager);
+            serviceConfig,
+            catalogManager,
+            tableManager,
+            optimizingService,
+            terminalManager,
+            tableService);
     RestCatalogService restCatalogService = new RestCatalogService(catalogManager, tableManager);
 
     httpServer =

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
@@ -48,12 +48,10 @@
 import org.apache.amoro.server.resource.OptimizerManager;
 import org.apache.amoro.server.resource.OptimizerThread;
 import org.apache.amoro.server.resource.QuotaProvider;
-import org.apache.amoro.server.table.DefaultTableServiceOld;
 import org.apache.amoro.server.table.MaintainedTableManager;
 import org.apache.amoro.server.table.RuntimeHandlerChain;
 import org.apache.amoro.server.table.TableRuntime;
 import org.apache.amoro.server.table.TableService;
-import org.apache.amoro.server.table.TableServiceOld;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList;
 import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -145,7 +143,7 @@ private void loadOptimizingQueues(List<TableRuntime> tableRuntimeMetaList) {
           List<TableRuntime> tableRuntimes = groupToTableRuntimes.remove(groupName);
           OptimizingQueue optimizingQueue =
               new OptimizingQueue(
-                  tableService,
+                  catalogManager,
                   group,
                   this,
                   planExecutor,
@@ -322,7 +320,7 @@ public void createResourceGroup(ResourceGroup resourceGroup) {
           doAs(ResourceMapper.class, mapper -> mapper.insertResourceGroup(resourceGroup));
           OptimizingQueue optimizingQueue =
               new OptimizingQueue(
-                  tableService,
+                  catalogManager,
                   resourceGroup,
                   this,
                   planExecutor,

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
@@ -44,7 +44,6 @@
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.manager.EventsManager;
 import org.apache.amoro.server.persistence.PersistentBase;
-import org.apache.amoro.server.table.TableServiceOld;
 import org.apache.amoro.server.table.internal.InternalTableCreator;
 import org.apache.amoro.server.table.internal.InternalTableHandler;
 import org.apache.amoro.server.table.internal.InternalTableManager;
@@ -348,7 +347,8 @@ public void deleteTable(Context ctx) {
               Boolean.parseBoolean(
                   Optional.ofNullable(ctx.req.getParameter("purgeRequested")).orElse("false"));
           org.apache.amoro.server.table.TableMetadata tableMetadata = handler.tableMetadata();
-          tableManager.dropTableMetadata(tableMetadata.getTableIdentifier().getIdentifier(), purge);
+          tableManager.dropTableMetadata(
+              tableMetadata.getTableIdentifier().getIdentifier().buildTableIdentifier(), purge);
           handler.dropTable(purge);
           return null;
         });

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
@@ -32,10 +32,8 @@
 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.MaintainedTableManager;
 import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableMetadata;
-import org.apache.amoro.server.table.TableServiceOld;
 import org.apache.amoro.server.utils.InternalTableUtil;
 import org.apache.amoro.shade.thrift.org.apache.thrift.TException;
 
@@ -91,7 +89,10 @@ public void createTableMeta(TableMeta tableMeta) {
     }
     ServerTableIdentifier identifier =
         ServerTableIdentifier.of(
-            tableMeta.getTableIdentifier(), TableFormat.valueOf(tableMeta.getFormat()));
+            tableMeta.getTableIdentifier().getCatalog(),
+            tableMeta.getTableIdentifier().getDatabase(),
+            tableMeta.getTableIdentifier().getTableName(),
+            TableFormat.valueOf(tableMeta.getFormat()));
     InternalCatalog catalog = catalogManager.getInternalCatalog(identifier.getCatalog());
     CatalogMeta catalogMeta = catalog.getMetadata();
     TableMetadata tableMetadata = new TableMetadata(identifier, tableMeta, catalogMeta);

diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
@@ -18,7 +18,9 @@
 
 package org.apache.amoro.server.catalog;
 
+import org.apache.amoro.AmoroTable;
 import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.table.TableIdentifier;
 
 import java.util.List;
 
@@ -90,4 +92,12 @@ public interface CatalogManager {
    * @param catalogMeta The CatalogMeta object representing the updated catalog information.
    */
   void updateCatalog(CatalogMeta catalogMeta);
+
+  /**
+   * load a table via server catalog.
+   *
+   * @param identifier managed table identifier
+   * @return managed table.
+   */
+  AmoroTable<?> loadTable(TableIdentifier identifier);
 }
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
@@ -18,6 +18,7 @@
 
 package org.apache.amoro.server.catalog;
 
+import org.apache.amoro.AmoroTable;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.exception.AlreadyExistsException;
@@ -31,6 +32,7 @@
 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.apache.amoro.table.TableIdentifier;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -195,6 +197,12 @@ public void updateCatalog(CatalogMeta catalogMeta) {
     LOG.info("Update catalog metadata: {}", catalogMeta.getCatalogName());
   }
 
+  @Override
+  public AmoroTable<?> loadTable(TableIdentifier identifier) {
+    ServerCatalog serverCatalog = getServerCatalog(identifier.getCatalog());
+    return serverCatalog.loadTable(identifier.getDatabase(), identifier.getTableName());
+  }
+
   private void validateCatalogUpdate(CatalogMeta oldMeta, CatalogMeta newMeta) {
     if (!oldMeta.getCatalogType().equals(newMeta.getCatalogType())) {
       throw new IllegalMetadataException("Cannot update catalog type");

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
@@ -22,7 +22,6 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.TableIDWithFormat;
 import org.apache.amoro.api.CatalogMeta;
-import org.apache.amoro.api.TableIdentifier;
 import org.apache.amoro.exception.AlreadyExistsException;
 import org.apache.amoro.exception.IllegalMetadataException;
 import org.apache.amoro.exception.ObjectNotExistsException;
@@ -32,6 +31,7 @@
 import org.apache.amoro.server.table.TableMetadata;
 import org.apache.amoro.server.table.internal.InternalTableCreator;
 import org.apache.amoro.server.table.internal.InternalTableHandler;
+import org.apache.amoro.table.TableIdentifier;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 
 import java.util.List;
@@ -101,11 +101,7 @@ public List<TableIDWithFormat> listTables(String database) {
             TableMetaMapper.class,
             mapper -> mapper.selectTableIdentifiersByDb(getMetadata().getCatalogName(), database))
         .stream()
-        .map(
-            sid ->
-                TableIDWithFormat.of(
-                    org.apache.amoro.table.TableIdentifier.of(sid.getIdentifier()),
-                    sid.getFormat()))
+        .map(sid -> TableIDWithFormat.of(sid.getIdentifier(), sid.getFormat()))
         .collect(Collectors.toList());
   }
 
@@ -115,11 +111,7 @@ public List<TableIDWithFormat> listTables() {
             TableMetaMapper.class,
             mapper -> mapper.selectTableIdentifiersByCatalog(getMetadata().getCatalogName()))
         .stream()
-        .map(
-            sid ->
-                TableIDWithFormat.of(
-                    org.apache.amoro.table.TableIdentifier.of(sid.getIdentifier()),
-                    sid.getFormat()))
+        .map(sid -> TableIDWithFormat.of(sid.getIdentifier(), sid.getFormat()))
         .collect(Collectors.toList());
   }
 

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
@@ -52,7 +52,7 @@
 import org.apache.amoro.server.dashboard.response.ErrorResponse;
 import org.apache.amoro.server.dashboard.utils.ParamSignatureCalculator;
 import org.apache.amoro.server.table.TableManager;
-import org.apache.amoro.server.table.TableServiceOld;
+import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
@@ -101,19 +101,24 @@ public DashboardServer(
       CatalogManager catalogManager,
       TableManager tableManager,
       DefaultOptimizingService optimizerManager,
-      TerminalManager terminalManager) {
+      TerminalManager terminalManager,
+      TableService tableService) {
     PlatformFileManager platformFileManager = new PlatformFileManager();
     this.catalogController = new CatalogController(catalogManager, platformFileManager);
     this.healthCheckController = new HealthCheckController();
     this.loginController = new LoginController(serviceConfig);
-    this.optimizerGroupController = new OptimizerGroupController(tableService, optimizerManager);
+    // TODO: remove table service from OptimizerGroupController
+    this.optimizerGroupController =
+        new OptimizerGroupController(tableManager, tableService, optimizerManager);
     this.optimizerController = new OptimizerController(optimizerManager);
     this.platformFileInfoController = new PlatformFileInfoController(platformFileManager);
     this.settingController = new SettingController(serviceConfig, optimizerManager);
     ServerTableDescriptor tableDescriptor =
         new ServerTableDescriptor(catalogManager, tableManager, serviceConfig);
+    // TODO: remove table service from TableController
     this.tableController =
-        new TableController(catalogManager, tableService, tableDescriptor, serviceConfig);
+        new TableController(
+            catalogManager, tableManager, 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
@@ -27,7 +27,6 @@
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.table.TableManager;
-import org.apache.amoro.server.table.TableServiceOld;
 import org.apache.amoro.table.descriptor.AmoroSnapshotsOfTable;
 import org.apache.amoro.table.descriptor.ConsumerInfo;
 import org.apache.amoro.table.descriptor.DDLInfo;

diff --git a/.../src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java b/.../src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java
@@ -34,8 +34,9 @@
 import org.apache.amoro.server.resource.ContainerMetadata;
 import org.apache.amoro.server.resource.OptimizerInstance;
 import org.apache.amoro.server.resource.ResourceContainers;
+import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableRuntime;
-import org.apache.amoro.server.table.TableServiceOld;
+import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
@@ -58,11 +59,15 @@ public class OptimizerGroupController {
   private static final Logger LOG = LoggerFactory.getLogger(OptimizerGroupController.class);
 
   private static final String ALL_GROUP = "all";
-  private final TableServiceOld tableService;
+  private final TableManager tableManager;
+  private final TableService tableService;
   private final DefaultOptimizingService optimizerManager;
 
   public OptimizerGroupController(
-      TableServiceOld tableService, DefaultOptimizingService optimizerManager) {
+      TableManager tableManager,
+      TableService tableService,
+      DefaultOptimizingService optimizerManager) {
+    this.tableManager = tableManager;
     this.tableService = tableService;
     this.optimizerManager = optimizerManager;
   }
@@ -102,7 +107,7 @@ public void getOptimizerTables(Context ctx) {
       statusCodes = null;
     }
     Pair<List<TableRuntimeMeta>, Integer> tableRuntimeBeans =
-        tableService.getTableRuntimes(
+        tableManager.queryTableRuntimeMetas(
             optimizerGroupUsedInDbFilter,
             dbFilterStr,
             tableFilterStr,

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
@@ -51,8 +51,9 @@
 import org.apache.amoro.server.dashboard.utils.AmsUtil;
 import org.apache.amoro.server.dashboard.utils.CommonUtil;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableRuntime;
-import org.apache.amoro.server.table.TableServiceOld;
+import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.shade.guava32.com.google.common.base.Function;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -101,7 +102,8 @@ public class TableController {
   private static final long UPGRADE_INFO_EXPIRE_INTERVAL = 60 * 60 * 1000;
 
   private final CatalogManager catalogManager;
-  private final TableServiceOld tableService;
+  private final TableManager tableManager;
+  private final TableService tableService;
   private final ServerTableDescriptor tableDescriptor;
   private final Configurations serviceConfig;
   private final ConcurrentHashMap<TableIdentifier, UpgradeRunningInfo> upgradeRunningInfo =
@@ -110,10 +112,12 @@ public class TableController {
 
   public TableController(
       CatalogManager catalogManager,
-      TableServiceOld tableService,
+      TableManager tableManager,
+      TableService tableService,
       ServerTableDescriptor tableDescriptor,
       Configurations serviceConfig) {
     this.catalogManager = catalogManager;
+    this.tableManager = tableManager;
     this.tableService = tableService;
     this.tableDescriptor = tableDescriptor;
     this.serviceConfig = serviceConfig;
@@ -150,7 +154,7 @@ public void getTableDetail(Context ctx) {
     TableSummary tableSummary = serverTableMeta.getTableSummary();
     Optional<ServerTableIdentifier> serverTableIdentifier =
         Optional.ofNullable(
-            tableService.getServerTableIdentifier(
+            tableManager.getServerTableIdentifier(
                 TableIdentifier.of(catalog, database, tableName).buildTableIdentifier()));
     if (serverTableIdentifier.isPresent()) {
       TableRuntime tableRuntime = tableService.getRuntime(serverTableIdentifier.get().getId());
@@ -679,7 +683,7 @@ public void cancelOptimizingProcess(Context ctx) {
     Preconditions.checkState(catalogManager.catalogExist(catalog), "invalid catalog!");
 
     ServerTableIdentifier serverTableIdentifier =
-        tableService.getServerTableIdentifier(
+        tableManager.getServerTableIdentifier(
             TableIdentifier.of(catalog, db, table).buildTableIdentifier());
     TableRuntime tableRuntime =
         serverTableIdentifier != null