Add support for dynamic refresh interval for table metadata refreshing in the TableRuntimeRefreshExecutor.

Author: 张文领

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java

@@ -56,19 +56,29 @@ protected boolean enabled(TableRuntime tableRuntime) {
   @Override
   protected long getNextExecutingTime(TableRuntime tableRuntime) {
     DefaultTableRuntime defaultTableRuntime = (DefaultTableRuntime) tableRuntime;
+
+    if (defaultTableRuntime.getOptimizingConfig().getRefreshTableAdaptiveEnabled()) {
+      long newInterval = defaultTableRuntime.getLatestRefreshInterval();
+      if (newInterval > 0) {
+        return newInterval;
+      }
+    }
+
     return Math.min(
         defaultTableRuntime.getOptimizingConfig().getMinorLeastInterval() * 4L / 5, interval);
   }
 
   private void tryEvaluatingPendingInput(DefaultTableRuntime tableRuntime, MixedTable table) {
     // only evaluate pending input when optimizing is enabled and in idle state
     OptimizingConfig optimizingConfig = tableRuntime.getOptimizingConfig();
-    if (optimizingConfig.isEnabled()
-        && tableRuntime.getOptimizingStatus().equals(OptimizingStatus.IDLE)) {
+    boolean optimizingEnabled = optimizingConfig.isEnabled();
+    if (optimizingEnabled && tableRuntime.getOptimizingStatus().equals(OptimizingStatus.IDLE)) {
 
       if (optimizingConfig.isMetadataBasedTriggerEnabled()
           && !MetadataBasedEvaluationEvent.isEvaluatingNecessary(
               optimizingConfig, table, tableRuntime.getLastPlanTime())) {
+        tableRuntime.setLatestEvaluatedNeedOptimizing(false);
+
         logger.debug(
             "{} optimizing is not necessary due to metadata based trigger",
             tableRuntime.getTableIdentifier());
@@ -87,8 +97,17 @@ private void tryEvaluatingPendingInput(DefaultTableRuntime tableRuntime, MixedTa
         tableRuntime.setPendingInput(pendingInput);
       } else {
         tableRuntime.optimizingNotNecessary();
+        tableRuntime.setTableSummary(evaluator.getPendingInput());
       }
-      tableRuntime.setTableSummary(evaluator.getPendingInput());
+    } else if (!optimizingEnabled) {
+      tableRuntime.setLatestEvaluatedNeedOptimizing(false);
+      logger.debug(
+          "{} optimizing is not enabled, skip evaluating pending input",
+          tableRuntime.getTableIdentifier());
+    } else {
+      tableRuntime.setLatestEvaluatedNeedOptimizing(true);
+      logger.debug(
+          "{} optimizing is processing or is in preparation", tableRuntime.getTableIdentifier());
     }
   }
 
@@ -129,9 +148,124 @@ public void execute(TableRuntime tableRuntime) {
           || (mixedTable.isUnkeyedTable()
               && lastOptimizedSnapshotId != defaultTableRuntime.getCurrentSnapshotId())) {
         tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+      } else {
+        logger.debug("{} optimizing is not necessary", defaultTableRuntime.getTableIdentifier());
+        defaultTableRuntime.setLatestEvaluatedNeedOptimizing(false);
+      }
+
+      // Update adaptive interval according to evaluating result.
+      if (defaultTableRuntime.getOptimizingConfig().getRefreshTableAdaptiveEnabled()) {
+        long newInterval = getAdaptiveExecutingInterval(defaultTableRuntime);
+        defaultTableRuntime.setLatestRefreshInterval(newInterval);
       }
+
     } catch (Throwable throwable) {
       logger.error("Refreshing table {} failed.", tableRuntime.getTableIdentifier(), throwable);
     }
   }
+
+  /**
+   * Calculate adaptive execution interval based on table optimization status.
+   *
+   * <p>Uses AIMD (Additive Increase Multiplicative Decrease) algorithm inspired by TCP congestion
+   * control:
+   *
+   * <ul>
+   *   <li>If table does not need to be optimized: additive increase - gradually extend interval to
+   *       reduce resource consumption
+   *   <li>If table needs optimization: multiplicative decrease - rapidly reduce interval for quick
+   *       response
+   * </ul>
+   *
+   * <p>Interval is bounded by [interval_min, interval_max] and kept in memory only (resets to
+   * interval_min on restart).
+   *
+   * @param tableRuntime The table runtime information containing current status and configuration
+   * @return The next execution interval in milliseconds
+   */
+  private long getAdaptiveExecutingInterval(DefaultTableRuntime tableRuntime) {
+    final long minInterval = interval;
+    final long maxInterval =
+        tableRuntime.getOptimizingConfig().getRefreshTableAdaptiveMaxInterval();
+    long currentInterval = tableRuntime.getLatestRefreshInterval();
+
+    // Initialize interval on first run or after restart
+    if (currentInterval == 0) {
+      currentInterval = minInterval;
+    }
+
+    // Determine whether table needs optimization
+    boolean needOptimizing = tableRuntime.getLatestEvaluatedNeedOptimizing();
+
+    long nextInterval;
+    if (needOptimizing) {
+      nextInterval = decreaseInterval(currentInterval, minInterval);
+      logger.debug(
+          "Table {} needs optimization, decreasing interval from {}ms to {}ms",
+          tableRuntime.getTableIdentifier(),
+          currentInterval,
+          nextInterval);
+    } else {
+      nextInterval = increaseInterval(tableRuntime, currentInterval, maxInterval);
+      logger.debug(
+          "Table {} does not need optimization, increasing interval from {}ms to {}ms",
+          tableRuntime.getTableIdentifier(),
+          currentInterval,
+          nextInterval);
+    }
+
+    return nextInterval;
+  }
+
+  /**
+   * Decrease interval when table needs optimization.
+   *
+   * <p>Uses multiplicative decrease (halving) inspired by TCP Fast Recovery algorithm for rapid
+   * response to table health issues.
+   *
+   * @param currentInterval Current refresh interval in milliseconds
+   * @param minInterval Minimum allowed interval in milliseconds
+   * @return New interval after decrease.
+   */
+  private long decreaseInterval(long currentInterval, long minInterval) {
+    long newInterval = currentInterval / 2;
+    long boundedInterval = Math.max(newInterval, minInterval);
+
+    if (newInterval < minInterval) {
+      logger.debug(
+          "Interval reached minimum boundary: attempted {}ms, capped at {}ms",
+          newInterval,
+          minInterval);
+    }
+
+    return boundedInterval;
+  }
+
+  /**
+   * Increase interval when table does not need optimization.
+   *
+   * <p>Uses additive increase inspired by TCP Congestion Avoidance algorithm for gradual and stable
+   * growth.
+   *
+   * @param tableRuntime The table runtime information containing configuration
+   * @param currentInterval Current refresh interval in milliseconds
+   * @param maxInterval Maximum allowed interval in milliseconds
+   * @return New interval after increase.
+   */
+  private long increaseInterval(
+      DefaultTableRuntime tableRuntime, long currentInterval, long maxInterval) {
+    long step = tableRuntime.getOptimizingConfig().getRefreshTableAdaptiveIncreaseStep();
+    long newInterval = currentInterval + step;
+    long boundedInterval = Math.min(newInterval, maxInterval);
+
+    if (newInterval > maxInterval) {
+      logger.debug(
+          "Interval reached maximum boundary: currentInterval is {}ms, attempted {}ms, capped at {}ms",
+          currentInterval,
+          newInterval,
+          maxInterval);
+    }
+
+    return boundedInterval;
+  }
 }

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

@@ -100,6 +100,8 @@ public class DefaultTableRuntime extends AbstractTableRuntime
   private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
   private final TableSummaryMetrics tableSummaryMetrics;
   private volatile long lastPlanTime;
+  private volatile long latestRefreshInterval = AmoroServiceConstants.INVALID_TIME;
+  private volatile boolean latestEvaluatedNeedOptimizing = true;
   private volatile OptimizingProcess optimizingProcess;
   private final List<TaskRuntime.TaskQuota> taskQuotas = new CopyOnWriteArrayList<>();
 
@@ -181,6 +183,22 @@ public void setLastPlanTime(long lastPlanTime) {
     this.lastPlanTime = lastPlanTime;
   }
 
+  public long getLatestRefreshInterval() {
+    return latestRefreshInterval;
+  }
+
+  public void setLatestRefreshInterval(long latestRefreshInterval) {
+    this.latestRefreshInterval = latestRefreshInterval;
+  }
+
+  public boolean getLatestEvaluatedNeedOptimizing() {
+    return this.latestEvaluatedNeedOptimizing;
+  }
+
+  public void setLatestEvaluatedNeedOptimizing(boolean latestEvaluatedNeedOptimizing) {
+    this.latestEvaluatedNeedOptimizing = latestEvaluatedNeedOptimizing;
+  }
+
   public OptimizingStatus getOptimizingStatus() {
     return OptimizingStatus.ofCode(getStatusCode());
   }
@@ -288,6 +306,7 @@ public void setPendingInput(AbstractOptimizingEvaluator.PendingInput pendingInpu
               summary.setTotalFileCount(pendingFileCount);
             })
         .commit();
+    this.latestEvaluatedNeedOptimizing = true;
   }
 
   public void setTableSummary(AbstractOptimizingEvaluator.PendingInput tableSummary) {
@@ -462,6 +481,8 @@ public void optimizingNotNecessary() {
               })
           .commit();
     }
+
+    this.latestEvaluatedNeedOptimizing = false;
   }
 
   public void beginCommitting() {

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

@@ -341,7 +341,22 @@ public static OptimizingConfig parseOptimizingConfig(Map<String, String> propert
             PropertyUtil.propertyAsLong(
                 properties,
                 TableProperties.SELF_OPTIMIZING_EVALUATION_FILE_SIZE_MSE_TOLERANCE,
-                TableProperties.SELF_OPTIMIZING_EVALUATION_FILE_SIZE_MSE_TOLERANCE_DEFAULT));
+                TableProperties.SELF_OPTIMIZING_EVALUATION_FILE_SIZE_MSE_TOLERANCE_DEFAULT))
+        .setRefreshTableAdaptiveEnabled(
+            PropertyUtil.propertyAsBoolean(
+                properties,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED_DEFAULT))
+        .setRefreshTableAdaptiveMaxInterval(
+            PropertyUtil.propertyAsLong(
+                properties,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL_DEFAULT))
+        .setRefreshTableAdaptiveIncreaseStep(
+            PropertyUtil.propertyAsLong(
+                properties,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_INCREASE_STEP,
+                TableProperties.SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_INCREASE_STEP_DEFAULT));
   }
 
   /**