Merge branch 'master' into optimizing-process

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

@@ -162,6 +162,12 @@ public class AmoroManagementConf {
           .defaultValue(60000L)
           .withDescription("Interval for refreshing table metadata.");
 
+  public static final ConfigOption<Integer> REFRESH_MAX_PENDING_PARTITIONS =
+      ConfigOptions.key("refresh-tables.max-pending-partition-count")
+          .intType()
+          .defaultValue(100)
+          .withDescription("Filters will not be used beyond that number of partitions");
+
   public static final ConfigOption<Long> BLOCKER_TIMEOUT =
       ConfigOptions.key("blocker.timeout")
           .longType()

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

@@ -26,12 +26,12 @@
 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.dashboard.DashboardServer;
 import org.apache.amoro.server.dashboard.JavalinJsonMapper;
 import org.apache.amoro.server.dashboard.response.ErrorResponse;
 import org.apache.amoro.server.dashboard.utils.AmsUtil;
 import org.apache.amoro.server.dashboard.utils.CommonUtil;
-import org.apache.amoro.server.exception.AmoroRuntimeException;
 import org.apache.amoro.server.manager.EventsManager;
 import org.apache.amoro.server.manager.MetricManager;
 import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;

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

@@ -29,14 +29,14 @@
 import org.apache.amoro.api.OptimizingTaskResult;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.exception.ForbiddenException;
+import org.apache.amoro.exception.IllegalTaskStateException;
+import org.apache.amoro.exception.ObjectNotExistsException;
+import org.apache.amoro.exception.PluginRetryAuthException;
+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.exception.ForbiddenException;
-import org.apache.amoro.server.exception.IllegalTaskStateException;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
-import org.apache.amoro.server.exception.PluginRetryAuthException;
-import org.apache.amoro.server.exception.TaskNotFoundException;
 import org.apache.amoro.server.optimizing.OptimizingQueue;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
 import org.apache.amoro.server.optimizing.TaskRuntime;

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

@@ -37,10 +37,10 @@
 import org.apache.amoro.ServerTableIdentifier;
 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.InternalCatalog;
 import org.apache.amoro.server.catalog.ServerCatalog;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
 import org.apache.amoro.server.manager.EventsManager;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.table.TableService;

amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java

@@ -23,9 +23,9 @@
 import org.apache.amoro.TableIDWithFormat;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.api.TableIdentifier;
-import org.apache.amoro.server.exception.AlreadyExistsException;
-import org.apache.amoro.server.exception.IllegalMetadataException;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
+import org.apache.amoro.exception.AlreadyExistsException;
+import org.apache.amoro.exception.IllegalMetadataException;
+import org.apache.amoro.exception.ObjectNotExistsException;
 import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
 import org.apache.amoro.server.persistence.mapper.TableBlockerMapper;
 import org.apache.amoro.server.persistence.mapper.TableMetaMapper;

amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java

@@ -27,12 +27,12 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.config.Configurations;
+import org.apache.amoro.exception.ObjectNotExistsException;
 import org.apache.amoro.formats.iceberg.IcebergTable;
 import org.apache.amoro.io.AuthenticatedFileIO;
 import org.apache.amoro.mixed.InternalMixedIcebergCatalog;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.RestCatalogService;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
 import org.apache.amoro.server.table.TableMetadata;
 import org.apache.amoro.server.table.internal.InternalIcebergCreator;
 import org.apache.amoro.server.table.internal.InternalIcebergHandler;

amoro-ams/src/main/java/org/apache/amoro/server/catalog/ServerCatalog.java

@@ -21,7 +21,7 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableIDWithFormat;
 import org.apache.amoro.api.CatalogMeta;
-import org.apache.amoro.server.exception.IllegalMetadataException;
+import org.apache.amoro.exception.IllegalMetadataException;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
 

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java

@@ -32,6 +32,8 @@
 import io.javalin.http.staticfiles.Location;
 import io.javalin.http.staticfiles.StaticFileConfig;
 import org.apache.amoro.config.Configurations;
+import org.apache.amoro.exception.ForbiddenException;
+import org.apache.amoro.exception.SignatureCheckException;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.DefaultOptimizingService;
 import org.apache.amoro.server.RestCatalogService;
@@ -48,8 +50,6 @@
 import org.apache.amoro.server.dashboard.controller.VersionController;
 import org.apache.amoro.server.dashboard.response.ErrorResponse;
 import org.apache.amoro.server.dashboard.utils.ParamSignatureCalculator;
-import org.apache.amoro.server.exception.ForbiddenException;
-import org.apache.amoro.server.exception.SignatureCheckException;
 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;

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/MixedAndIcebergTableDescriptor.java

@@ -723,10 +723,13 @@ private ServerTableMeta getServerTableMeta(MixedTable table) {
                 .map(item -> buildColumnInfoFromPartitionSpec(table.spec().schema(), item))
                 .collect(Collectors.toList()));
       }
+    } else {
+      serverTableMeta.setPkList(
+          serverTableMeta.getSchema().stream()
+              .filter(s -> table.schema().identifierFieldNames().contains(s.getField()))
+              .collect(Collectors.toList()));
     }
-    if (serverTableMeta.getPkList() == null) {
-      serverTableMeta.setPkList(new ArrayList<>());
-    }
+
     return serverTableMeta;
   }
 

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/utils/CommonUtil.java

@@ -19,7 +19,7 @@
 package org.apache.amoro.server.dashboard.utils;
 
 import io.javalin.http.Context;
-import org.apache.amoro.server.exception.SignatureCheckException;
+import org.apache.amoro.exception.SignatureCheckException;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.net.telnet.TelnetClient;
 import org.slf4j.Logger;

amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractPluginManager.java

@@ -19,9 +19,9 @@
 package org.apache.amoro.server.manager;
 
 import org.apache.amoro.ActivePlugin;
+import org.apache.amoro.exception.AlreadyExistsException;
+import org.apache.amoro.exception.LoadingPluginException;
 import org.apache.amoro.server.Environments;
-import org.apache.amoro.server.exception.AlreadyExistsException;
-import org.apache.amoro.server.exception.LoadingPluginException;
 import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList;

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/KeyedTableCommit.java

@@ -24,13 +24,13 @@
 import org.apache.amoro.api.CommitMetaProducer;
 import org.apache.amoro.data.DataFileType;
 import org.apache.amoro.data.PrimaryKeyedFile;
+import org.apache.amoro.exception.OptimizingCommitException;
 import org.apache.amoro.hive.utils.TableTypeUtil;
 import org.apache.amoro.op.OverwriteBaseFiles;
 import org.apache.amoro.op.SnapshotSummary;
 import org.apache.amoro.optimizing.RewriteFilesInput;
 import org.apache.amoro.optimizing.RewriteFilesOutput;
 import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.exception.OptimizingCommitException;
 import org.apache.amoro.table.MixedTable;
 import org.apache.amoro.utils.ContentFiles;
 import org.apache.amoro.utils.MixedTableUtil;

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java

@@ -22,10 +22,10 @@
 import org.apache.amoro.OptimizerProperties;
 import org.apache.amoro.ServerTableIdentifier;
 import org.apache.amoro.api.OptimizingTaskId;
+import org.apache.amoro.exception.OptimizingClosedException;
 import org.apache.amoro.optimizing.RewriteFilesInput;
 import org.apache.amoro.resource.ResourceGroup;
 import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.exception.OptimizingClosedException;
 import org.apache.amoro.server.manager.MetricManager;
 import org.apache.amoro.server.optimizing.plan.OptimizingPlanner;
 import org.apache.amoro.server.persistence.PersistentBase;
@@ -666,12 +666,13 @@ private void cancelTasks() {
     }
 
     private void loadTaskRuntimes(OptimizingProcess optimizingProcess) {
-      List<TaskRuntime<RewriteStageTask>> taskRuntimes =
-          getAs(
-              OptimizingMapper.class,
-              mapper ->
-                  mapper.selectTaskRuntimes(tableRuntime.getTableIdentifier().getId(), processId));
       try {
+        List<TaskRuntime<RewriteStageTask>> taskRuntimes =
+            getAs(
+                OptimizingMapper.class,
+                mapper ->
+                    mapper.selectTaskRuntimes(
+                        tableRuntime.getTableIdentifier().getId(), processId));
         Map<Integer, RewriteFilesInput> inputs = TaskFilesPersistence.loadTaskInputs(processId);
         taskRuntimes.forEach(
             taskRuntime -> {

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java

@@ -22,10 +22,10 @@
 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.server.AmoroServiceConstants;
-import org.apache.amoro.server.exception.IllegalTaskStateException;
-import org.apache.amoro.server.exception.OptimizingClosedException;
-import org.apache.amoro.server.exception.TaskRuntimeException;
 import org.apache.amoro.server.persistence.StatedPersistentBase;
 import org.apache.amoro.server.persistence.mapper.OptimizingMapper;
 import org.apache.amoro.server.resource.OptimizerThread;
@@ -281,7 +281,7 @@ public void accept(Status targetStatus) {
         throw new OptimizingClosedException(taskId.getProcessId());
       }
       if (!getNext().contains(targetStatus)) {
-        throw new IllegalTaskStateException(taskId, status, targetStatus);
+        throw new IllegalTaskStateException(taskId, status.name(), targetStatus.name());
       }
       status = targetStatus;
     }

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java

@@ -23,6 +23,7 @@
 
 import org.apache.amoro.api.CommitMetaProducer;
 import org.apache.amoro.data.FileNameRules;
+import org.apache.amoro.exception.OptimizingCommitException;
 import org.apache.amoro.hive.HMSClientPool;
 import org.apache.amoro.hive.table.SupportHive;
 import org.apache.amoro.hive.utils.HivePartitionUtil;
@@ -33,7 +34,6 @@
 import org.apache.amoro.optimizing.RewriteFilesOutput;
 import org.apache.amoro.properties.HiveTableProperties;
 import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.exception.OptimizingCommitException;
 import org.apache.amoro.server.utils.IcebergTableUtil;
 import org.apache.amoro.table.MixedTable;
 import org.apache.amoro.table.UnkeyedTable;

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java

@@ -62,15 +62,18 @@ public class OptimizingEvaluator {
   protected final MixedTable mixedTable;
   protected final TableRuntime tableRuntime;
   protected final TableSnapshot currentSnapshot;
+  protected final int maxPendingPartitions;
   protected boolean isInitialized = false;
 
   protected Map<String, PartitionEvaluator> needOptimizingPlanMap = Maps.newHashMap();
   protected Map<String, PartitionEvaluator> partitionPlanMap = Maps.newHashMap();
 
-  public OptimizingEvaluator(TableRuntime tableRuntime, MixedTable table) {
+  public OptimizingEvaluator(
+      TableRuntime tableRuntime, MixedTable table, int maxPendingPartitions) {
     this.tableRuntime = tableRuntime;
     this.mixedTable = table;
     this.currentSnapshot = IcebergTableUtil.getSnapshot(table, tableRuntime);
+    this.maxPendingPartitions = maxPendingPartitions;
   }
 
   public TableRuntime getTableRuntime() {
@@ -137,6 +140,7 @@ private void initPartitionPlans(TableFileScanHelper tableFileScanHelper) {
     needOptimizingPlanMap.putAll(
         partitionPlanMap.entrySet().stream()
             .filter(entry -> entry.getValue().isNecessary())
+            .limit(maxPendingPartitions)
             .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())));
   }
 

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingPlanner.java

@@ -65,7 +65,7 @@ public OptimizingPlanner(
       MixedTable table,
       double availableCore,
       long maxInputSizePerThread) {
-    super(tableRuntime, table);
+    super(tableRuntime, table, Integer.MAX_VALUE);
     this.partitionFilter =
         tableRuntime.getPendingInput() == null
             ? Expressions.alwaysTrue()
@@ -175,11 +175,11 @@ public List<RewriteStageTask> planTasks() {
       tasks.addAll(partitionPlan.splitTasks((int) (actualInputSize / avgThreadCost)));
     }
     if (!tasks.isEmpty()) {
-      if (evaluators.stream()
-          .anyMatch(evaluator -> evaluator.getOptimizingType() == OptimizingType.FULL)) {
+      if (actualPartitionPlans.stream()
+          .anyMatch(plan -> plan.getOptimizingType() == OptimizingType.FULL)) {
         optimizingType = OptimizingType.FULL;
-      } else if (evaluators.stream()
-          .anyMatch(evaluator -> evaluator.getOptimizingType() == OptimizingType.MAJOR)) {
+      } else if (actualPartitionPlans.stream()
+          .anyMatch(plan -> plan.getOptimizingType() == OptimizingType.MAJOR)) {
         optimizingType = OptimizingType.MAJOR;
       } else {
         optimizingType = OptimizingType.MINOR;

amoro-ams/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java

@@ -18,8 +18,8 @@
 
 package org.apache.amoro.server.persistence;
 
-import org.apache.amoro.server.exception.AmoroRuntimeException;
-import org.apache.amoro.server.exception.PersistenceException;
+import org.apache.amoro.exception.AmoroRuntimeException;
+import org.apache.amoro.exception.PersistenceException;
 import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.ibatis.session.TransactionIsolationLevel;

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

@@ -29,16 +29,16 @@
 import org.apache.amoro.api.TableIdentifier;
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.exception.AlreadyExistsException;
+import org.apache.amoro.exception.BlockerConflictException;
+import org.apache.amoro.exception.IllegalMetadataException;
+import org.apache.amoro.exception.ObjectNotExistsException;
+import org.apache.amoro.exception.PersistenceException;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.catalog.CatalogBuilder;
 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.exception.AlreadyExistsException;
-import org.apache.amoro.server.exception.BlockerConflictException;
-import org.apache.amoro.server.exception.IllegalMetadataException;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
-import org.apache.amoro.server.exception.PersistenceException;
 import org.apache.amoro.server.manager.MetricManager;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
 import org.apache.amoro.server.persistence.StatedPersistentBase;

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

@@ -77,7 +77,8 @@ public void setup(TableManager tableManager, Configurations conf) {
         new TableRuntimeRefreshExecutor(
             tableManager,
             conf.getInteger(AmoroManagementConf.REFRESH_TABLES_THREAD_COUNT),
-            conf.getLong(AmoroManagementConf.REFRESH_TABLES_INTERVAL));
+            conf.getLong(AmoroManagementConf.REFRESH_TABLES_INTERVAL),
+            conf.getInteger(AmoroManagementConf.REFRESH_MAX_PENDING_PARTITIONS));
     if (conf.getBoolean(AmoroManagementConf.AUTO_CREATE_TAGS_ENABLED)) {
       this.tagsAutoCreatingExecutor =
           new TagsAutoCreatingExecutor(

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

@@ -31,10 +31,13 @@ public class TableRuntimeRefreshExecutor extends BaseTableExecutor {
 
   // 1 minutes
   private final long interval;
+  private final int maxPendingPartitions;
 
-  public TableRuntimeRefreshExecutor(TableManager tableRuntimes, int poolSize, long interval) {
+  public TableRuntimeRefreshExecutor(
+      TableManager tableRuntimes, int poolSize, long interval, int maxPendingPartitions) {
     super(tableRuntimes, poolSize);
     this.interval = interval;
+    this.maxPendingPartitions = maxPendingPartitions;
   }
 
   @Override
@@ -48,7 +51,8 @@ protected long getNextExecutingTime(TableRuntime tableRuntime) {
 
   private void tryEvaluatingPendingInput(TableRuntime tableRuntime, MixedTable table) {
     if (tableRuntime.isOptimizingEnabled() && !tableRuntime.getOptimizingStatus().isProcessing()) {
-      OptimizingEvaluator evaluator = new OptimizingEvaluator(tableRuntime, table);
+      OptimizingEvaluator evaluator =
+          new OptimizingEvaluator(tableRuntime, table, maxPendingPartitions);
       if (evaluator.isNecessary()) {
         OptimizingEvaluator.PendingInput pendingInput = evaluator.getOptimizingPendingInput();
         logger.debug(

amoro-ams/src/main/java/org/apache/amoro/server/utils/PreconditionUtils.java

@@ -18,8 +18,8 @@
 
 package org.apache.amoro.server.utils;
 
-import org.apache.amoro.server.exception.AlreadyExistsException;
-import org.apache.amoro.server.exception.ObjectNotExistsException;
+import org.apache.amoro.exception.AlreadyExistsException;
+import org.apache.amoro.exception.ObjectNotExistsException;
 
 public class PreconditionUtils {