apache
diff --git a/‎amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java‎
Lines changed: 270 additions & 11 deletions b/‎amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java‎
Lines changed: 270 additions & 11 deletions
diff --git a/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java‎
Lines changed: 42 additions & 0 deletions b/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java‎
Lines changed: 42 additions & 0 deletions
diff --git a/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java‎
Lines changed: 16 additions & 0 deletions b/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java‎
Lines changed: 16 additions & 0 deletions
diff --git a/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java‎
Lines changed: 52 additions & 0 deletions b/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java‎
Lines changed: 52 additions & 0 deletions
diff --git a/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java‎
Lines changed: 206 additions & 0 deletions b/‎amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java‎
Lines changed: 206 additions & 0 deletions
@@ -38,6 +38,7 @@
 import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.manager.MetricManager;
 import org.apache.amoro.server.optimizing.TaskRuntime.Status;
+import org.apache.amoro.server.optimizing.dra.DynamicAllocationState;
 import org.apache.amoro.server.persistence.OptimizingProcessState;
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.server.persistence.TaskFilesPersistence;
@@ -418,6 +419,47 @@ public List<TaskRuntime<?>> collectTasks(Predicate<TaskRuntime<?>> predicate) {
         .collect(Collectors.toList());
   }
 
+  /**
+   * Snapshot the demand-side load of this queue for dynamic allocation: busy threads, serviceable
+   * PLANNED tasks (quota-mode aware, see {@link DynamicAllocationState#serviceablePlannedCount}),
+   * and PENDING tables. PENDING tables are observable with zero optimizers, which makes them the
+   * only scale-up signal on a cold group where nothing polls and planning never runs.
+   */
+  public DynamicAllocationState.GroupLoad collectDynamicAllocationLoad() {
+    Map<Long, Integer> plannedByTable = Maps.newHashMap();
+    Map<Long, Integer> occupiedByTable = Maps.newHashMap();
+    int busyThreads = 0;
+    for (TaskRuntime<?> task : collectTasks()) {
+      if (DynamicAllocationState.occupiesThread(task.getStatus())) {
+        busyThreads++;
+        occupiedByTable.merge(task.getTableId(), 1, Integer::sum);
+      } else if (task.getStatus() == Status.PLANNED) {
+        plannedByTable.merge(task.getTableId(), 1, Integer::sum);
+      }
+    }
+    Map<Long, Double> targetQuotaByTable = Maps.newHashMap();
+    int pendingTables = 0;
+    for (DefaultTableRuntime tableRuntime : scheduler.snapshotTableRuntimes()) {
+      targetQuotaByTable.put(
+          tableRuntime.getTableIdentifier().getId(),
+          tableRuntime.getOptimizingConfig().getTargetQuota());
+      if (tableRuntime.getOptimizingStatus() == OptimizingStatus.PENDING) {
+        pendingTables++;
+      }
+    }
+    List<DynamicAllocationState.TableDemand> demands = Lists.newArrayList();
+    plannedByTable.forEach(
+        (tableId, planned) ->
+            demands.add(
+                new DynamicAllocationState.TableDemand(
+                    planned,
+                    // Unknown table (racing removal): default to proportional, counting in full.
+                    targetQuotaByTable.getOrDefault(tableId, 1.0),
+                    occupiedByTable.getOrDefault(tableId, 0))));
+    return new DynamicAllocationState.GroupLoad(
+        busyThreads, DynamicAllocationState.serviceablePlannedCount(demands), pendingTables);
+  }
+
   public void retryTask(TaskRuntime<?> taskRuntime) {
     findProcess(taskRuntime.getTaskId()).resetTask((TaskRuntime<RewriteStageTask>) taskRuntime);
   }
 
@@ -28,9 +28,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.ServiceLoader;
@@ -146,4 +148,18 @@ public void removeTable(DefaultTableRuntime tableRuntime) {
   Map<ServerTableIdentifier, DefaultTableRuntime> getTableRuntimeMap() {
     return tableRuntimeMap;
   }
+
+  /**
+   * Copy the current table runtimes under {@code tableLock}. {@code tableRuntimeMap} is a plain
+   * {@link HashMap} whose canonical accesses all hold the lock, so callers on other threads (e.g.
+   * the dynamic-allocation scale keeper) must iterate a snapshot instead of the live map.
+   */
+  public List<DefaultTableRuntime> snapshotTableRuntimes() {
+    tableLock.lock();
+    try {
+      return new ArrayList<>(tableRuntimeMap.values());
+    } finally {
+      tableLock.unlock();
+    }
+  }
 }
@@ -42,6 +42,7 @@ public class DynamicAllocationConfig {
   private final boolean enabled;
   private final Integer minParallelism;
   private final Integer maxParallelism;
+  private final int executorParallelism;
   private final Duration schedulerBacklogTimeout;
   private final Duration sustainedBacklogTimeout;
   private final Duration executorIdleTimeout;
@@ -54,6 +55,7 @@ private DynamicAllocationConfig(
       boolean enabled,
       Integer minParallelism,
       Integer maxParallelism,
+      int executorParallelism,
       Duration schedulerBacklogTimeout,
       Duration sustainedBacklogTimeout,
       Duration executorIdleTimeout,
@@ -64,6 +66,7 @@ private DynamicAllocationConfig(
     this.enabled = enabled;
     this.minParallelism = minParallelism;
     this.maxParallelism = maxParallelism;
+    this.executorParallelism = executorParallelism;
     this.schedulerBacklogTimeout = schedulerBacklogTimeout;
     this.sustainedBacklogTimeout = sustainedBacklogTimeout;
     this.executorIdleTimeout = executorIdleTimeout;
@@ -88,12 +91,19 @@ public static DynamicAllocationConfig parse(ResourceGroup group) {
         PropertyUtil.propertyAsNullableInt(
             properties, OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM);
 
+    int executorParallelism =
+        PropertyUtil.propertyAsInt(
+            properties,
+            OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+            OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM_DEFAULT);
+
     return new DynamicAllocationConfig(
         group.getName(),
         group.getContainer(),
         enabled,
         minParallelism,
         maxParallelism,
+        executorParallelism,
         parseDuration(
             properties,
             OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
@@ -269,6 +279,44 @@ public void validate() {
               maxParallelism,
               OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT));
     }
+    if (executorParallelism < 1) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Resource group:%s '%s'(%d) must be >= 1.",
+              groupName,
+              OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+              executorParallelism));
+    }
+    // A single executorParallelism-thread instance is the scaling unit; if it alone exceeds
+    // max-parallelism, scale-up could never create anything, leaving a silent no-op group.
+    if (executorParallelism > maxParallelism) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Resource group:%s '%s'(%d) must not exceed '%s'(%d).",
+              groupName,
+              OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+              executorParallelism,
+              OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+              maxParallelism));
+    }
+    // The floor is satisfied in executor-parallelism-thread instance units; if covering it would
+    // already exceed max-parallelism, the group would silently sit below its floor forever.
+    int floorThreads =
+        (minParallelism + executorParallelism - 1) / executorParallelism * executorParallelism;
+    if (floorThreads > maxParallelism) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Resource group:%s '%s'(%d) is not reachable in '%s'(%d) units: covering the floor "
+                  + "requires %d threads, exceeding '%s'(%d).",
+              groupName,
+              OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
+              minParallelism,
+              OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+              executorParallelism,
+              floorThreads,
+              OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+              maxParallelism));
+    }
     Duration idleMin =
         ConfigHelpers.TimeUtils.parseDuration(
             OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN);
@@ -340,6 +388,10 @@ public int getMaxParallelism() {
     return maxParallelism;
   }
 
+  public int getExecutorParallelism() {
+    return executorParallelism;
+  }
+
   public Duration getSchedulerBacklogTimeout() {
     return schedulerBacklogTimeout;
   }
 
@@ -0,0 +1,206 @@
+/*
+ * 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.optimizing.dra;
+
+import org.apache.amoro.server.optimizing.TaskRuntime;
+
+import java.util.Collection;
+
+/**
+ * Per-group scale-up decision state for dynamic allocation (AIP-5): the backlog timer, the
+ * scale-out cadence, and the exponential ramp. {@link #computeScaleUp} is driven by injected inputs
+ * (loads, config, time), so the decision logic is deterministic and testable without a live
+ * optimizing queue; the static demand-accounting helpers are pure functions.
+ */
+public final class DynamicAllocationState {
+
+  /** When demand was first observed; {@code -1} while there is no demand. */
+  private long backlogSinceMs = -1;
+
+  /** Earliest time the next scale-out may happen; {@code -1} before the first one. */
+  private long nextAllowedAddMs = -1;
+
+  /** Instances to add in the next immediate-demand round (1, 2, 4, 8 ...). */
+  private int rampInstances = 1;
+
+  /**
+   * Decide how many executor-parallelism-thread optimizer instances to create in this round.
+   *
+   * <p>Ordered checks: the {@code min-parallelism} floor is enforced immediately (no timing gate);
+   * immediate demand ({@code busy + serviceable > effective}) scales exponentially, clamped to the
+   * actual need (Spark semantics: the ramp resets when the clamp binds, and a round with no demand
+   * resets it too); future demand (pending tables while every thread is busy — including the
+   * zero-optimizer cold start, where nothing polls and planning never runs) adds a single probe
+   * instance, because pending tables are not quantified demand before planning. Demand must persist
+   * for {@code scheduler-backlog-timeout} before the first scale-out; subsequent ones are spaced by
+   * {@code sustained-backlog-timeout}. The {@code max-parallelism} cap always wins.
+   *
+   * @return the number of instances of {@code executor-parallelism} threads to create, {@code >= 0}
+   */
+  public int computeScaleUp(
+      int effectiveThreads,
+      int busyThreads,
+      int serviceablePlanned,
+      int pendingTables,
+      DynamicAllocationConfig config,
+      long nowMs) {
+    int k = config.getExecutorParallelism();
+    int allowedInstances = Math.max(0, (config.getMaxParallelism() - effectiveThreads) / k);
+
+    int minParallelism = config.getMinParallelism();
+    if (effectiveThreads < minParallelism) {
+      // A floor deficit (optimizers died or the group is new) invalidates any demand-phase
+      // state: after recovery, demand must re-prove backlog persistence instead of firing
+      // through a stale gate with a stale ramp.
+      backlogSinceMs = -1;
+      nextAllowedAddMs = -1;
+      rampInstances = 1;
+      int neededInstances = ceilDiv(minParallelism - effectiveThreads, k);
+      return Math.min(neededInstances, allowedInstances);
+    }
+
+    int actionableNeed = Math.max(busyThreads + serviceablePlanned - effectiveThreads, 0);
+    boolean futureDemand = pendingTables > 0 && busyThreads >= effectiveThreads;
+    if (actionableNeed <= 0 && !futureDemand) {
+      backlogSinceMs = -1;
+      nextAllowedAddMs = -1;
+      rampInstances = 1;
+      return 0;
+    }
+
+    if (backlogSinceMs < 0) {
+      backlogSinceMs = nowMs;
+      nextAllowedAddMs = -1;
+    }
+    long gate =
+        nextAllowedAddMs >= 0
+            ? nextAllowedAddMs
+            : backlogSinceMs + config.getSchedulerBacklogTimeout().toMillis();
+    if (nowMs < gate) {
+      return 0;
+    }
+
+    int add;
+    if (actionableNeed > 0) {
+      int wantInstances = ceilDiv(actionableNeed, k);
+      add = Math.min(Math.min(wantInstances, rampInstances), allowedInstances);
+      if (add <= 0) {
+        return 0;
+      }
+      // Spark semantics: keep doubling only while the ramp is the binding constraint; once the
+      // actual need clamps the add, a grown ramp is no longer justified by demand.
+      rampInstances = wantInstances > rampInstances ? rampInstances * 2 : 1;
+    } else {
+      add = Math.min(1, allowedInstances);
+      if (add <= 0) {
+        return 0;
+      }
+      rampInstances = 1;
+    }
+    nextAllowedAddMs = nowMs + config.getSustainedBacklogTimeout().toMillis();
+    return add;
+  }
+
+  private static int ceilDiv(int value, int divisor) {
+    return (value + divisor - 1) / divisor;
+  }
+
+  /** Snapshot of a group's current load, the demand-side inputs of {@link #computeScaleUp}. */
+  public static class GroupLoad {
+    private final int busyThreads;
+    private final int serviceablePlanned;
+    private final int pendingTables;
+
+    public GroupLoad(int busyThreads, int serviceablePlanned, int pendingTables) {
+      this.busyThreads = busyThreads;
+      this.serviceablePlanned = serviceablePlanned;
+      this.pendingTables = pendingTables;
+    }
+
+    public int getBusyThreads() {
+      return busyThreads;
+    }
+
+    public int getServiceablePlanned() {
+      return serviceablePlanned;
+    }
+
+    public int getPendingTables() {
+      return pendingTables;
+    }
+  }
+
+  /** Per-table demand snapshot consumed by {@link #serviceablePlannedCount(Collection)}. */
+  public static class TableDemand {
+    private final int plannedCount;
+    private final double targetQuota;
+    private final int occupiedThreads;
+
+    public TableDemand(int plannedCount, double targetQuota, int occupiedThreads) {
+      this.plannedCount = plannedCount;
+      this.targetQuota = targetQuota;
+      this.occupiedThreads = occupiedThreads;
+    }
+  }
+
+  /**
+   * Count the PLANNED tasks that adding optimizer capacity could actually drain.
+   *
+   * <p>A table with a proportional quota ({@code targetQuota <= 1}) is limited to {@code
+   * ceil(targetQuota * availableCore)} threads, so scaling up raises its limit and the whole
+   * backlog is serviceable. A table with an absolute quota ({@code > 1}) has a fixed thread limit
+   * that scaling cannot raise, so only its currently free slots are serviceable.
+   */
+  public static int serviceablePlannedCount(Collection<TableDemand> demands) {
+    int total = 0;
+    for (TableDemand demand : demands) {
+      if (demand.targetQuota <= 1) {
+        total += demand.plannedCount;
+      } else {
+        int freeSlots = Math.max(0, (int) demand.targetQuota - demand.occupiedThreads);
+        total += Math.min(demand.plannedCount, freeSlots);
+      }
+    }
+    return total;
+  }
+
+  /**
+   * Whether a task in this status occupies an optimizer thread. A thread is occupied from
+   * assignment ({@code SCHEDULED}, set by {@code pollTask}) until the task terminates; counting
+   * only {@code ACKED} would overestimate headroom during the poll-to-ack window.
+   */
+  public static boolean occupiesThread(TaskRuntime.Status status) {
+    return status == TaskRuntime.Status.SCHEDULED || status == TaskRuntime.Status.ACKED;
+  }
+
+  /**
+   * Whether the group is bottlenecked on planning rather than on thread capacity: threads sit idle
+   * while tables wait as PENDING and no PLANNED tasks materialize (planning is serialized by {@code
+   * optimizer.max-planning-parallelism}). Scaling out in this state would only add more idle
+   * threads, so the condition is surfaced as a warning instead of a scale-out. A cold group (zero
+   * threads) is the future-demand case, not a planning bottleneck.
+   */
+  public static boolean isPlanningBound(
+      int effectiveThreads, int busyThreads, int serviceablePlanned, int pendingTables) {
+    return effectiveThreads > 0
+        && busyThreads < effectiveThreads
+        && serviceablePlanned == 0
+        && pendingTables > 0;
+  }
+}