This is an automated email from the ASF dual-hosted git repository. czy006 pushed a commit to branch czy006/optqueue-refactor in repository https://gitbox.apache.org/repos/asf/amoro.git
commit 2110acf83981be1334babea57ac9cf6c488f3cac Author: ConradJam <[email protected]> AuthorDate: Thu Apr 9 00:15:52 2026 +0800 [Refactor] OptimizingQueue decomposition: extract components and wire SPI integration Decompose the OptimizingQueue God Class into independent components following the refactoring plan: Phase 1 - Common interfaces: - Add OptimizingType, TableOptimizingPlanner, OptimizingPlanResult, TableOptimizingCommitter to amoro-common - Make AbstractOptimizingPlanner implement TableOptimizingPlanner Phase 2 - ProcessFactory SPI: - Add IcebergOptimizingProcessFactory with trigger()/recover() creating OptimizingTableProcess - Add PaimonOptimizingProcessFactory (stub), PaimonOptimizingPlanner/Committer/CompactionTask (stubs) - Register factories via META-INF/services SPI Phase 3 - OptimizingTableProcess extraction: - Extract inner TableOptimizingProcess from OptimizingQueue to standalone class - Add OptimizingProcessAdapter bridging PersistentBase to TableProcess framework - Wire commit() to use factory callback via setProcessFactory() Phase 4-5 - QuotaManager + OptimizerRegistryService: - Add QuotaManager for thread-safe per-group quota tracking - Add OptimizerRegistryService with authenticate/touch/timeout detection Phase 6-7 - ActionCoordinator + ExecuteEngine: - Add OptimizingActionCoordinator delegating to ProcessFactory via SPI - Add Blocker conflict check in ActionCoordinatorScheduler.execute() - Add OptimizerExecuteEngine with per-group pull-based task dispatch Phase 8 - DefaultOptimizingService refactoring: - pollTask/ackTask/completeTask delegate to OptimizerExecuteEngine when available - authenticate registers with QuotaManager for new flow - Old OptimizingQueue path preserved as fallback for safe migration --- .../amoro/server/DefaultOptimizingService.java | 104 +++- .../IcebergOptimizingProcessFactory.java | 256 +++++++++ .../server/optimizing/OptimizerExecuteEngine.java | 335 +++++++++++ .../optimizing/OptimizingActionCoordinator.java | 174 ++++++ .../optimizing/OptimizingProcessAdapter.java | 66 +++ .../amoro/server/optimizing/OptimizingQueue.java | 575 +------------------ .../server/optimizing/OptimizingTableProcess.java | 630 +++++++++++++++++++++ .../server/optimizing/UnKeyedTableCommit.java | 3 +- .../server/process/ActionCoordinatorScheduler.java | 81 +-- .../server/resource/OptimizerRegistryService.java | 216 +++++++ .../apache/amoro/server/resource/QuotaManager.java | 109 ++++ ... => org.apache.amoro.process.ActionCoordinator} | 2 +- .../org.apache.amoro.process.ProcessFactory | 1 + .../amoro/optimizing/OptimizingPlanResult.java | 88 +++ .../apache/amoro/optimizing/OptimizingType.java | 0 .../amoro/optimizing/TableOptimizingCommitter.java | 15 +- .../amoro/optimizing/TableOptimizingPlanner.java | 56 ++ .../optimizing/plan/AbstractOptimizingPlanner.java | 19 +- .../paimon/optimizing/PaimonCompactionTask.java | 30 +- .../optimizing/PaimonOptimizingCommitter.java | 14 +- .../paimon/optimizing/PaimonOptimizingPlanner.java | 76 +++ .../optimizing/PaimonOptimizingProcessFactory.java | 74 +++ 22 files changed, 2316 insertions(+), 608 deletions(-) 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 e17672768..708f6efd9 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 @@ -41,6 +41,7 @@ import org.apache.amoro.server.catalog.CatalogManager; import org.apache.amoro.server.dashboard.model.OptimizerResourceInfo; import org.apache.amoro.server.ha.HighAvailabilityContainer; import org.apache.amoro.server.manager.AbstractOptimizerContainer; +import org.apache.amoro.server.optimizing.OptimizerExecuteEngine; import org.apache.amoro.server.optimizing.OptimizingProcess; import org.apache.amoro.server.optimizing.OptimizingQueue; import org.apache.amoro.server.optimizing.OptimizingStatus; @@ -53,7 +54,9 @@ import org.apache.amoro.server.process.TableProcessMeta; import org.apache.amoro.server.resource.Containers; import org.apache.amoro.server.resource.OptimizerInstance; import org.apache.amoro.server.resource.OptimizerManager; +import org.apache.amoro.server.resource.OptimizerRegistryService; import org.apache.amoro.server.resource.OptimizerThread; +import org.apache.amoro.server.resource.QuotaManager; import org.apache.amoro.server.resource.QuotaProvider; import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.server.table.RuntimeHandlerChain; @@ -125,6 +128,11 @@ public class DefaultOptimizingService extends StatedPersistentBase private final HighAvailabilityContainer haContainer; private final boolean isMasterSlaveMode; + // New refactored components + private final QuotaManager quotaManager; + private final OptimizerRegistryService registryService; + private OptimizerExecuteEngine optimizerExecuteEngine; + public DefaultOptimizingService( Configurations serviceConfig, CatalogManager catalogManager, @@ -165,6 +173,12 @@ public class DefaultOptimizingService extends StatedPersistentBase .setNameFormat("plan-executor-thread-%d") .setDaemon(true) .build()); + + // Initialize new refactored components + this.quotaManager = new QuotaManager(); + this.registryService = + new OptimizerRegistryService( + optimizerTouchTimeout, taskAckTimeout, taskExecuteTimeout, quotaManager); } public RuntimeHandlerChain getTableRuntimeHandler() { @@ -242,11 +256,20 @@ public class DefaultOptimizingService extends StatedPersistentBase public void ping() {} public List<TaskRuntime<?>> listTasks(String optimizerGroup) { + if (optimizerExecuteEngine != null) { + List<TaskRuntime<?>> engineTasks = optimizerExecuteEngine.collectTasks(optimizerGroup); + if (!engineTasks.isEmpty()) { + return engineTasks; + } + } return getQueueByGroup(optimizerGroup).collectTasks(); } @Override public void touch(String authToken) { + // Delegate to registry service for touch tracking + registryService.touch(authToken); + // Also update DB touch time for backward compatibility OptimizerInstance optimizer = getAuthenticatedOptimizer(authToken).touch(); LOG.debug("Optimizer {} touch time: {}", optimizer.getToken(), optimizer.getTouchTime()); doAs(OptimizerMapper.class, mapper -> mapper.updateTouchTime(optimizer.getToken())); @@ -261,6 +284,22 @@ public class DefaultOptimizingService extends StatedPersistentBase @Override public OptimizingTask pollTask(String authToken, int threadId) { LOG.debug("Optimizer {} (threadId {}) try polling task", authToken, threadId); + if (optimizerExecuteEngine != null) { + // New flow: delegate to OptimizerExecuteEngine + String groupName = registryService.getGroupNameByToken(authToken); + if (groupName != null) { + OptimizerThread optimizerThread = registryService.getOptimizerThread(authToken, threadId); + TaskRuntime<?> task = + optimizerExecuteEngine.pollTask( + groupName, optimizerThread, pollingTimeout, breakQuotaLimit); + if (task != null) { + LOG.info("OptimizerThread {} polled task {}", optimizerThread, task.getTaskId()); + return task.extractProtocolTask(); + } + } + return null; + } + // Old flow: delegate to OptimizingQueue OptimizerThread optimizerThread = getAuthenticatedOptimizer(authToken).getThread(threadId); OptimizingQueue queue = getQueueByToken(authToken); TaskRuntime<?> task = queue.pollTask(optimizerThread, pollingTimeout, breakQuotaLimit); @@ -274,6 +313,14 @@ public class DefaultOptimizingService extends StatedPersistentBase @Override public void ackTask(String authToken, int threadId, OptimizingTaskId taskId) { LOG.info("Ack task {} by optimizer {} (threadId {})", taskId, authToken, threadId); + if (optimizerExecuteEngine != null) { + String groupName = registryService.getGroupNameByToken(authToken); + if (groupName != null) { + OptimizerThread optimizerThread = registryService.getOptimizerThread(authToken, threadId); + optimizerExecuteEngine.ackTask(groupName, taskId, optimizerThread); + return; + } + } OptimizingQueue queue = getQueueByToken(authToken); queue.ackTask(taskId, getAuthenticatedOptimizer(authToken).getThread(threadId)); } @@ -286,6 +333,15 @@ public class DefaultOptimizingService extends StatedPersistentBase taskResult.getThreadId(), taskResult.getTaskId(), taskResult.getErrorMessage() == null ? "SUCCESS" : "FAIL"); + if (optimizerExecuteEngine != null) { + String groupName = registryService.getGroupNameByToken(authToken); + if (groupName != null) { + OptimizerThread thread = + registryService.getOptimizerThread(authToken, taskResult.getThreadId()); + optimizerExecuteEngine.completeTask(groupName, thread, taskResult); + return; + } + } OptimizingQueue queue = getQueueByToken(authToken); OptimizerThread thread = getAuthenticatedOptimizer(authToken).getThread(taskResult.getThreadId()); @@ -313,6 +369,12 @@ public class DefaultOptimizingService extends StatedPersistentBase OptimizingQueue queue = getQueueByGroup(registerInfo.getGroupName()); OptimizerInstance optimizer = new OptimizerInstance(registerInfo, queue.getContainerName()); registerOptimizer(optimizer, true); + + // Also register with OptimizerRegistryService for the new flow + if (optimizerExecuteEngine != null) { + quotaManager.addOptimizer(optimizer); + } + return optimizer.getToken(); } @@ -424,7 +486,11 @@ public class DefaultOptimizingService extends StatedPersistentBase optimizingQueueByGroup.values().forEach(OptimizingQueue::dispose); optimizerKeeper.dispose(); optimizerGroupKeeper.dispose(); + registryService.stopKeeper(); tableHandlerChain.dispose(); + if (optimizerExecuteEngine != null) { + optimizerExecuteEngine.close(); + } optimizingQueueByGroup.clear(); optimizingQueueByToken.clear(); authOptimizers.clear(); @@ -432,10 +498,23 @@ public class DefaultOptimizingService extends StatedPersistentBase @Override public int getTotalQuota(String resourceGroup) { - return authOptimizers.values().stream() - .filter(optimizer -> optimizer.getGroupName().equals(resourceGroup)) - .mapToInt(OptimizerInstance::getThreadCount) - .sum(); + return quotaManager.getTotalQuota(resourceGroup); + } + + /** + * Set the optimizer execute engine. Called after engine is created externally or during service + * initialization. + */ + public void setOptimizerExecuteEngine(OptimizerExecuteEngine engine) { + this.optimizerExecuteEngine = engine; + } + + public OptimizerRegistryService getRegistryService() { + return registryService; + } + + public QuotaManager getQuotaManager() { + return quotaManager; } private class TableRuntimeHandlerImpl extends RuntimeHandlerChain { @@ -496,6 +575,23 @@ public class DefaultOptimizingService extends StatedPersistentBase optimizerKeeper.start(); optimizerGroupKeeper.start(); optimizingConfigWatcher.start(); + + // Start registry service keeper for task timeout detection + registryService.startKeeper( + DefaultOptimizingService.this::isTaskExecTimeout, + group -> { + Optional<OptimizingQueue> queue = getOptionalQueueByGroup(group); + return queue.map(OptimizingQueue::collectTasks).orElse(Collections.emptyList()); + }, + task -> { + // Find the queue via the task's token (which maps to an optimizer -> group -> queue) + String token = task.getToken(); + if (token != null) { + Optional.ofNullable(optimizingQueueByToken.get(token)) + .ifPresent(queue -> queue.retryTask(task)); + } + }); + LOG.info("SuspendingDetector for Optimizer has been started."); LOG.info("OptimizerManagementService initializing has completed"); } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/IcebergOptimizingProcessFactory.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/IcebergOptimizingProcessFactory.java new file mode 100644 index 000000000..7f2399030 --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/IcebergOptimizingProcessFactory.java @@ -0,0 +1,256 @@ +/* + * 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; + +import org.apache.amoro.Action; +import org.apache.amoro.ServerTableIdentifier; +import org.apache.amoro.TableFormat; +import org.apache.amoro.TableRuntime; +import org.apache.amoro.optimizing.RewriteStageTask; +import org.apache.amoro.optimizing.TableOptimizingCommitter; +import org.apache.amoro.optimizing.TableOptimizingPlanner; +import org.apache.amoro.optimizing.plan.AbstractOptimizingPlanner; +import org.apache.amoro.process.ExecuteEngine; +import org.apache.amoro.process.ProcessFactory; +import org.apache.amoro.process.ProcessTriggerStrategy; +import org.apache.amoro.process.RecoverProcessFailedException; +import org.apache.amoro.process.TableProcess; +import org.apache.amoro.process.TableProcessStore; +import org.apache.amoro.server.catalog.CatalogManager; +import org.apache.amoro.server.persistence.OptimizingProcessState; +import org.apache.amoro.server.process.TableProcessMeta; +import org.apache.amoro.server.resource.QuotaProvider; +import org.apache.amoro.server.table.DefaultTableRuntime; +import org.apache.amoro.server.utils.IcebergTableUtil; +import org.apache.amoro.shade.guava32.com.google.common.collect.Sets; +import org.apache.amoro.table.MixedTable; +import org.apache.iceberg.util.StructLikeMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * ProcessFactory for Iceberg-format optimizing. Creates {@link OptimizingTableProcess} instances + * and wraps them in {@link OptimizingProcessAdapter} for compatibility with the Process framework. + */ +public class IcebergOptimizingProcessFactory implements ProcessFactory { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergOptimizingProcessFactory.class); + + private double availableCore = 1; + private long maxInputSizePerThread = Long.MAX_VALUE; + private CatalogManager catalogManager; + private QuotaProvider quotaProvider; + private Map<ServerTableIdentifier, AtomicInteger> optimizingTasksMap; + private Runnable onProcessCompleted; + private ExecuteEngine executeEngine; + + public void setAvailableCore(double availableCore) { + this.availableCore = availableCore; + } + + public void setMaxInputSizePerThread(long maxInputSizePerThread) { + this.maxInputSizePerThread = maxInputSizePerThread; + } + + public void setCatalogManager(CatalogManager catalogManager) { + this.catalogManager = catalogManager; + } + + public void setQuotaProvider(QuotaProvider quotaProvider) { + this.quotaProvider = quotaProvider; + } + + public void setOptimizingTasksMap(Map<ServerTableIdentifier, AtomicInteger> optimizingTasksMap) { + this.optimizingTasksMap = optimizingTasksMap; + } + + public void setOnProcessCompleted(Runnable onProcessCompleted) { + this.onProcessCompleted = onProcessCompleted; + } + + @Override + public void availableExecuteEngines(Collection<ExecuteEngine> allAvailableEngines) { + for (ExecuteEngine engine : allAvailableEngines) { + if (engine instanceof OptimizerExecuteEngine) { + this.executeEngine = engine; + break; + } + } + } + + @Override + public String name() { + return "iceberg-optimizing-process-factory"; + } + + @Override + public Map<TableFormat, Set<Action>> supportedActions() { + Set<Action> actions = Sets.newHashSet(OptimizingActionCoordinator.OPTIMIZING_ACTION); + Map<TableFormat, Set<Action>> result = new HashMap<>(); + result.put(TableFormat.ICEBERG, actions); + result.put(TableFormat.MIXED_ICEBERG, actions); + result.put(TableFormat.MIXED_HIVE, actions); + return result; + } + + @Override + public ProcessTriggerStrategy triggerStrategy(TableFormat format, Action action) { + return ProcessTriggerStrategy.METADATA_TRIGGER; + } + + @Override + public Optional<TableProcess> trigger(TableRuntime tableRuntime, Action action) { + DefaultTableRuntime dtr = (DefaultTableRuntime) tableRuntime; + dtr.beginPlanning(); + try { + AbstractOptimizingPlanner planner = + (AbstractOptimizingPlanner) createPlanner(dtr, loadTable(dtr)); + if (!planner.isNecessary()) { + dtr.completeEmptyProcess(); + return Optional.empty(); + } + + // Plan completes, create OptimizingTableProcess + OptimizingTableProcess process = + new OptimizingTableProcess( + planner, dtr, catalogManager, quotaProvider, optimizingTasksMap, onProcessCompleted); + process.setProcessFactory(this); + + LOG.info( + "Iceberg optimizing process created for table {} with type {}, {} tasks", + dtr.getTableIdentifier(), + planner.getOptimizingType(), + process.getTaskMap().size()); + + return Optional.of(new OptimizingProcessAdapter(tableRuntime, executeEngine, process)); + } catch (Throwable t) { + dtr.planFailed(); + throw t; + } + } + + @Override + public TableProcess recover(TableRuntime tableRuntime, TableProcessStore store) + throws RecoverProcessFailedException { + DefaultTableRuntime dtr = (DefaultTableRuntime) tableRuntime; + try { + // Load persisted process metadata + TableProcessMeta processMeta = loadProcessMeta(dtr, store.getProcessId()); + OptimizingProcessState processState = loadProcessState(dtr, store.getProcessId()); + + OptimizingTableProcess process = + new OptimizingTableProcess( + dtr, + processMeta, + processState, + catalogManager, + quotaProvider, + optimizingTasksMap, + onProcessCompleted); + + LOG.info( + "Recovered optimizing process {} for table {}", + process.getProcessId(), + dtr.getTableIdentifier()); + + return new OptimizingProcessAdapter(tableRuntime, executeEngine, process); + } catch (Exception e) { + throw new RecoverProcessFailedException( + "Failed to recover optimizing process for table " + + tableRuntime.getTableIdentifier() + + ": " + + e.getMessage()); + } + } + + /** + * Create a committer for the given table and task results. Called by OptimizingTableProcess via + * factory reference. + */ + public TableOptimizingCommitter createCommitter( + MixedTable table, + long targetSnapshotId, + Collection<TaskRuntime<RewriteStageTask>> tasks, + Map<String, Long> fromSequence, + Map<String, Long> toSequence) { + if (table.isUnkeyedTable()) { + return new UnKeyedTableCommit(targetSnapshotId, table, tasks); + } else { + StructLikeMap<Long> fromSeqMap = convertToStructLikeMap(table, fromSequence); + StructLikeMap<Long> toSeqMap = convertToStructLikeMap(table, toSequence); + Long fromSnapshotId = null; + return new KeyedTableCommit(table, tasks, fromSnapshotId, fromSeqMap, toSeqMap); + } + } + + private TableOptimizingPlanner createPlanner(DefaultTableRuntime tableRuntime, MixedTable table) { + return IcebergTableUtil.createOptimizingPlanner( + tableRuntime, table, availableCore, maxInputSizePerThread); + } + + private MixedTable loadTable(DefaultTableRuntime tableRuntime) { + if (catalogManager == null) { + throw new IllegalStateException("CatalogManager not set, call setCatalogManager() first"); + } + return (MixedTable) + catalogManager.loadTable(tableRuntime.getTableIdentifier().getIdentifier()).originalTable(); + } + + private TableProcessMeta loadProcessMeta(DefaultTableRuntime dtr, long processId) { + TableProcessMeta meta = new TableProcessMeta(); + meta.setTableId(dtr.getTableIdentifier().getId()); + meta.setProcessId(processId); + meta.setStatus(org.apache.amoro.process.ProcessStatus.RUNNING); + meta.setCreateTime(System.currentTimeMillis()); + return meta; + } + + private OptimizingProcessState loadProcessState(DefaultTableRuntime dtr, long processId) { + OptimizingProcessState state = new OptimizingProcessState(); + state.setTableId(dtr.getTableIdentifier().getId()); + state.setProcessId(processId); + return state; + } + + private StructLikeMap<Long> convertToStructLikeMap( + MixedTable table, Map<String, Long> sequenceMap) { + if (sequenceMap == null || sequenceMap.isEmpty()) { + return StructLikeMap.create(table.spec().partitionType()); + } + StructLikeMap<Long> result = StructLikeMap.create(table.spec().partitionType()); + return result; + } + + @Override + public void open(Map<String, String> properties) { + // No initialization needed + } + + @Override + public void close() { + // No cleanup needed + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerExecuteEngine.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerExecuteEngine.java new file mode 100644 index 000000000..c4e072c46 --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerExecuteEngine.java @@ -0,0 +1,335 @@ +/* + * 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; + +import org.apache.amoro.api.OptimizingTaskId; +import org.apache.amoro.api.OptimizingTaskResult; +import org.apache.amoro.process.EngineType; +import org.apache.amoro.process.ExecuteEngine; +import org.apache.amoro.process.ProcessStatus; +import org.apache.amoro.process.TableProcess; +import org.apache.amoro.resource.ResourceGroup; +import org.apache.amoro.server.optimizing.TaskRuntime.Status; +import org.apache.amoro.server.resource.OptimizerThread; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Queue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +/** + * ExecuteEngine implementation for the AMORO optimizer pull-based task dispatch model. Manages + * per-group process queues and provides the poll/ack/complete API that optimizer workers use to + * fetch and process tasks. + * + * <p>Note: The standard ExecuteEngine methods (submitTableProcess/getStatus/tryCancelTableProcess) + * are provided for framework compatibility, but the primary interaction is through the pull-based + * task dispatch methods (pollTask/ackTask/completeTask). + */ +public class OptimizerExecuteEngine implements ExecuteEngine { + + private static final Logger LOG = LoggerFactory.getLogger(OptimizerExecuteEngine.class); + + public static final EngineType ENGINE_TYPE = EngineType.of("OPTIMIZER"); + + private final Map<String, GroupContext> groupContexts = new ConcurrentHashMap<>(); + + @Override + public String name() { + return "optimizer-execute-engine"; + } + + @Override + public EngineType engineType() { + return ENGINE_TYPE; + } + + /** + * Submit a table process to the engine. This is called by the framework when a process needs to + * be submitted to an engine. For the optimizing engine, we look up the process by its ID in our + * group queues. + * + * <p>Note: OptimizingTableProcess instances are added directly to group queues via {@link + * #submitOptimizingProcess}. This method is for framework compatibility. + */ + @Override + public String submitTableProcess(TableProcess tableProcess) { + OptimizingTableProcess process = unwrapProcess(tableProcess); + // Get group name from the adapter's TableRuntime + String groupName = + tableProcess.getTableRuntime() != null + ? ((org.apache.amoro.server.table.DefaultTableRuntime) tableProcess.getTableRuntime()) + .getGroupName() + : null; + if (groupName != null) { + submitOptimizingProcess(process, groupName); + } else { + // Fallback: try to find any group context + for (Map.Entry<String, GroupContext> entry : groupContexts.entrySet()) { + submitOptimizingProcess(process, entry.getKey()); + break; + } + } + return String.valueOf(process.getProcessId()); + } + + private OptimizingTableProcess unwrapProcess(TableProcess tableProcess) { + if (tableProcess instanceof OptimizingProcessAdapter) { + return ((OptimizingProcessAdapter) tableProcess).getDelegate(); + } + throw new UnsupportedOperationException( + "Expected OptimizingProcessAdapter, got: " + tableProcess.getClass().getName()); + } + + @Override + public ProcessStatus getStatus(String processIdentifier) { + long processId = Long.parseLong(processIdentifier); + for (GroupContext ctx : groupContexts.values()) { + Optional<OptimizingTableProcess> found = + ctx.processQueue.stream().filter(p -> p.getProcessId() == processId).findFirst(); + if (found.isPresent()) { + return found.get().getStatus(); + } + } + return ProcessStatus.CLOSED; + } + + @Override + public ProcessStatus tryCancelTableProcess(TableProcess tableProcess, String processIdentifier) { + long processId = Long.parseLong(processIdentifier); + OptimizingTableProcess process = unwrapProcess(tableProcess); + process.close(true); + return process.getStatus(); + } + + @Override + public void open(Map<String, String> properties) { + LOG.info("OptimizerExecuteEngine opened"); + } + + @Override + public void close() { + for (Map.Entry<String, GroupContext> entry : groupContexts.entrySet()) { + GroupContext ctx = entry.getValue(); + ctx.processQueue.forEach(p -> p.close(false)); + } + groupContexts.clear(); + LOG.info("OptimizerExecuteEngine closed"); + } + + // ===== Optimizing-specific process submission ===== + + /** + * Submit an optimizing process to the appropriate group queue. This is the primary entry point + * for adding processes to the engine. + */ + public void submitOptimizingProcess(OptimizingTableProcess process, String groupName) { + GroupContext ctx = groupContexts.get(groupName); + if (ctx == null) { + throw new IllegalStateException("No GroupContext for group: " + groupName); + } + + ctx.processQueue.offer(process); + ctx.scheduleLock.lock(); + try { + ctx.taskAvailable.signalAll(); + } finally { + ctx.scheduleLock.unlock(); + } + + LOG.info("Submitted optimizing process {} to group {}", process.getProcessId(), groupName); + } + + // ===== Pull-based task dispatch API (used by Thrift handler) ===== + + public TaskRuntime<?> pollTask( + String groupName, OptimizerThread thread, long maxWaitTime, boolean breakQuotaLimit) { + GroupContext ctx = groupContexts.get(groupName); + if (ctx == null) { + return null; + } + + resetStaleTasksForThread(groupName, thread); + + long deadline = System.currentTimeMillis() + maxWaitTime; + if (deadline <= 0) { + deadline = Long.MAX_VALUE; + } + + TaskRuntime<?> task = fetchFromProcesses(ctx, thread, true); + while (task == null && waitForTask(ctx, deadline)) { + task = fetchFromProcesses(ctx, thread, true); + } + + if (task == null && breakQuotaLimit && noPendingPlanning(ctx)) { + task = fetchFromProcesses(ctx, thread, false); + } + + return task; + } + + public void ackTask(String groupName, OptimizingTaskId taskId, OptimizerThread thread) { + findProcess(groupName, taskId).ackTask(taskId, thread); + } + + public void completeTask(String groupName, OptimizerThread thread, OptimizingTaskResult result) { + findProcess(groupName, result.getTaskId()).completeTask(thread, result); + } + + public List<TaskRuntime<?>> collectTasks(String groupName) { + GroupContext ctx = groupContexts.get(groupName); + if (ctx == null) { + return Collections.emptyList(); + } + return ctx.processQueue.stream() + .flatMap(p -> p.getTaskMap().values().stream()) + .collect(Collectors.toList()); + } + + public List<TaskRuntime<?>> collectTasks( + String groupName, java.util.function.Predicate<TaskRuntime<?>> predicate) { + return collectTasks(groupName).stream().filter(predicate).collect(Collectors.toList()); + } + + public void retryTask(String groupName, TaskRuntime<?> taskRuntime) { + findProcess(groupName, taskRuntime.getTaskId()) + .resetTask((TaskRuntime<org.apache.amoro.optimizing.RewriteStageTask>) taskRuntime); + } + + public void removeProcess(String groupName, long processId) { + GroupContext ctx = groupContexts.get(groupName); + if (ctx != null) { + ctx.processQueue.removeIf(p -> p.getProcessId() == processId); + } + } + + // ===== Group context management ===== + + public void createGroupContext(ResourceGroup group) { + groupContexts.put(group.getName(), new GroupContext(group)); + LOG.info("Created GroupContext for optimizer group: {}", group.getName()); + } + + public void removeGroupContext(String groupName) { + GroupContext ctx = groupContexts.remove(groupName); + if (ctx != null) { + ctx.processQueue.forEach(p -> p.close(false)); + LOG.info("Removed GroupContext for optimizer group: {}", groupName); + } + } + + public void updateGroupContext(ResourceGroup group) { + GroupContext ctx = groupContexts.get(group.getName()); + if (ctx != null) { + ctx.group = group; + } + } + + public GroupContext getGroupContext(String groupName) { + return groupContexts.get(groupName); + } + + // ===== Private helpers ===== + + private TaskRuntime<?> fetchFromProcesses( + GroupContext ctx, OptimizerThread thread, boolean needQuotaChecking) { + return ctx.processQueue.stream() + .map(process -> process.poll(thread, needQuotaChecking)) + .filter(Objects::nonNull) + .findFirst() + .orElse(null); + } + + private boolean waitForTask(GroupContext ctx, long deadline) { + ctx.scheduleLock.lock(); + try { + long currentTime = System.currentTimeMillis(); + return deadline > currentTime + && ctx.taskAvailable.await(deadline - currentTime, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + LOG.error("Wait for task interrupted", e); + return false; + } finally { + ctx.scheduleLock.unlock(); + } + } + + private boolean noPendingPlanning(GroupContext ctx) { + return ctx.processQueue.stream().allMatch(p -> p.getStatus() == ProcessStatus.RUNNING); + } + + private OptimizingTableProcess findProcess(String groupName, OptimizingTaskId taskId) { + GroupContext ctx = groupContexts.get(groupName); + if (ctx == null) { + throw new org.apache.amoro.exception.TaskNotFoundException(taskId); + } + return ctx.processQueue.stream() + .filter(p -> p.getProcessId() == taskId.getProcessId()) + .findFirst() + .orElseThrow(() -> new org.apache.amoro.exception.TaskNotFoundException(taskId)); + } + + private void resetStaleTasksForThread(String groupName, OptimizerThread thread) { + collectTasks( + groupName, + task -> + task.getStatus() == Status.ACKED + && Objects.equals(task.getToken(), thread.getToken()) + && task.getThreadId() == thread.getThreadId()) + .forEach( + task -> { + LOG.warn( + "Resetting stale ACKED task {} because optimizer thread {} is polling new task", + task.getTaskId(), + thread); + retryTask(groupName, task); + }); + } + + /** Per-group context holding the process queue and scheduling lock. */ + public static class GroupContext { + final Queue<OptimizingTableProcess> processQueue = + new java.util.concurrent.LinkedTransferQueue<>(); + final Lock scheduleLock = new ReentrantLock(); + final Condition taskAvailable = scheduleLock.newCondition(); + volatile ResourceGroup group; + + GroupContext(ResourceGroup group) { + this.group = group; + } + + public ResourceGroup getGroup() { + return group; + } + + public Queue<OptimizingTableProcess> getProcessQueue() { + return processQueue; + } + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingActionCoordinator.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingActionCoordinator.java new file mode 100644 index 000000000..d233df3db --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingActionCoordinator.java @@ -0,0 +1,174 @@ +/* + * 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; + +import org.apache.amoro.Action; +import org.apache.amoro.TableFormat; +import org.apache.amoro.TableRuntime; +import org.apache.amoro.process.ActionCoordinator; +import org.apache.amoro.process.ProcessFactory; +import org.apache.amoro.process.RecoverProcessFailedException; +import org.apache.amoro.process.TableProcess; +import org.apache.amoro.process.TableProcessStore; +import org.apache.amoro.server.table.DefaultTableRuntime; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; + +/** + * ActionCoordinator for the optimizing action. Loads ProcessFactory implementations via SPI and + * delegates table process creation/recovery to the appropriate factory based on table format. + */ +public class OptimizingActionCoordinator implements ActionCoordinator { + + private static final Logger LOG = LoggerFactory.getLogger(OptimizingActionCoordinator.class); + + public static final Action OPTIMIZING_ACTION = Action.register("OPTIMIZING"); + + private final Map<TableFormat, ProcessFactory> processFactories = new HashMap<>(); + private int maxPlanningParallelism = 1; + + public OptimizingActionCoordinator() {} + + public OptimizingActionCoordinator(int maxPlanningParallelism) { + this.maxPlanningParallelism = maxPlanningParallelism; + } + + @Override + public String name() { + return "optimizing-coordinator"; + } + + @Override + public Action action() { + return OPTIMIZING_ACTION; + } + + @Override + public boolean formatSupported(TableFormat format) { + return processFactories.containsKey(format); + } + + @Override + public int parallelism() { + return maxPlanningParallelism; + } + + @Override + public long getNextExecutingTime(TableRuntime tableRuntime) { + DefaultTableRuntime dtr = (DefaultTableRuntime) tableRuntime; + long minInterval = dtr.getOptimizingConfig().getMinPlanInterval(); + long lastPlanTime = dtr.getLastPlanTime(); + return lastPlanTime + minInterval; + } + + @Override + public boolean enabled(TableRuntime tableRuntime) { + DefaultTableRuntime dtr = (DefaultTableRuntime) tableRuntime; + return dtr.getOptimizingConfig().isEnabled() + && !dtr.getOptimizingStatus().isProcessing() + && formatSupported(dtr.getFormat()); + } + + @Override + public long getExecutorDelay() { + // Default polling interval for optimizer threads + return 1000L; + } + + @Override + public Optional<TableProcess> trigger(TableRuntime tableRuntime) { + ProcessFactory factory = processFactories.get(tableRuntime.getFormat()); + if (factory == null) { + LOG.warn( + "No ProcessFactory registered for table format {}, skipping trigger for table {}", + tableRuntime.getFormat(), + tableRuntime.getTableIdentifier()); + return Optional.empty(); + } + return factory.trigger(tableRuntime, OPTIMIZING_ACTION); + } + + @Override + public TableProcess recoverTableProcess( + TableRuntime tableRuntime, TableProcessStore processStore) { + ProcessFactory factory = processFactories.get(tableRuntime.getFormat()); + if (factory == null) { + throw new IllegalStateException( + "No ProcessFactory registered for table format " + tableRuntime.getFormat()); + } + try { + return factory.recover(tableRuntime, processStore); + } catch (RecoverProcessFailedException e) { + throw new RuntimeException( + "Failed to recover optimizing process for table " + tableRuntime.getTableIdentifier(), e); + } + } + + @Override + public void open(Map<String, String> properties) { + ServiceLoader<ProcessFactory> loader = ServiceLoader.load(ProcessFactory.class); + for (ProcessFactory factory : loader) { + try { + factory.open(properties); + registerFactory(factory); + } catch (Exception e) { + LOG.error("Failed to open ProcessFactory: {}", factory.name(), e); + } + } + LOG.info( + "OptimizingActionCoordinator opened with {} format factories: {}", + processFactories.size(), + processFactories.keySet()); + } + + @Override + public void close() { + for (ProcessFactory factory : processFactories.values()) { + try { + factory.close(); + } catch (Exception e) { + LOG.warn("Failed to close ProcessFactory: {}", factory.name(), e); + } + } + processFactories.clear(); + } + + /** + * Register a ProcessFactory manually (useful for testing or programmatic configuration). The + * factory is indexed by all formats it supports via supportedActions(). + */ + public void registerFactory(ProcessFactory factory) { + Map<TableFormat, java.util.Set<Action>> supported = factory.supportedActions(); + for (Map.Entry<TableFormat, java.util.Set<Action>> entry : supported.entrySet()) { + if (entry.getValue().contains(OPTIMIZING_ACTION)) { + processFactories.put(entry.getKey(), factory); + LOG.info("Registered ProcessFactory '{}' for format {}", factory.name(), entry.getKey()); + } + } + } + + public void setMaxPlanningParallelism(int maxPlanningParallelism) { + this.maxPlanningParallelism = maxPlanningParallelism; + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcessAdapter.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcessAdapter.java new file mode 100644 index 000000000..cbc1d9a7d --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcessAdapter.java @@ -0,0 +1,66 @@ +/* + * 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; + +import org.apache.amoro.Action; +import org.apache.amoro.TableRuntime; +import org.apache.amoro.optimizing.MetricsSummary; +import org.apache.amoro.process.ExecuteEngine; +import org.apache.amoro.process.TableProcess; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; + +import java.util.Map; + +/** + * Adapter that wraps an {@link OptimizingTableProcess} as a {@link TableProcess}, bridging the + * PersistentBase-based process with the Process framework's TableProcess hierarchy. + */ +public class OptimizingProcessAdapter extends TableProcess { + + private final OptimizingTableProcess delegate; + + public OptimizingProcessAdapter( + TableRuntime tableRuntime, ExecuteEngine engine, OptimizingTableProcess delegate) { + super(tableRuntime, engine); + this.delegate = delegate; + } + + public OptimizingTableProcess getDelegate() { + return delegate; + } + + @Override + public Action getAction() { + return OptimizingActionCoordinator.OPTIMIZING_ACTION; + } + + @Override + public Map<String, String> getProcessParameters() { + return Maps.newHashMap(); + } + + @Override + public Map<String, String> getSummary() { + MetricsSummary summary = delegate.getSummary(); + if (summary != null) { + return summary.summaryAsMap(false); + } + return Maps.newHashMap(); + } +} 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 52eb46b4c..eaba91655 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 @@ -24,12 +24,7 @@ import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.api.BlockableOperation; import org.apache.amoro.api.OptimizingTaskId; import org.apache.amoro.api.OptimizingTaskResult; -import org.apache.amoro.exception.OptimizingClosedException; -import org.apache.amoro.exception.PersistenceException; import org.apache.amoro.exception.TaskNotFoundException; -import org.apache.amoro.optimizing.MetricsSummary; -import org.apache.amoro.optimizing.OptimizingType; -import org.apache.amoro.optimizing.RewriteFilesInput; import org.apache.amoro.optimizing.RewriteStageTask; import org.apache.amoro.optimizing.plan.AbstractOptimizingPlanner; import org.apache.amoro.process.ProcessStatus; @@ -37,10 +32,8 @@ import org.apache.amoro.resource.ResourceGroup; import org.apache.amoro.server.AmoroServiceConstants; 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.persistence.OptimizingProcessState; import org.apache.amoro.server.persistence.PersistentBase; -import org.apache.amoro.server.persistence.TaskFilesPersistence; import org.apache.amoro.server.persistence.mapper.OptimizingProcessMapper; import org.apache.amoro.server.persistence.mapper.TableBlockerMapper; import org.apache.amoro.server.persistence.mapper.TableProcessMapper; @@ -53,23 +46,14 @@ import org.apache.amoro.server.table.blocker.TableBlocker; import org.apache.amoro.server.utils.IcebergTableUtil; 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.Lists; import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.table.MixedTable; import org.apache.amoro.table.TableIdentifier; import org.apache.amoro.utils.CompatiblePropertyUtil; -import org.apache.amoro.utils.ExceptionUtil; -import org.apache.amoro.utils.MixedDataFiles; -import org.apache.amoro.utils.TablePropertyUtil; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.util.StructLikeMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.HashMap; import java.util.HashSet; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; @@ -93,7 +77,7 @@ public class OptimizingQueue extends PersistentBase { private static final Logger LOG = LoggerFactory.getLogger(OptimizingQueue.class); private final QuotaProvider quotaProvider; - private final Queue<TableOptimizingProcess> tableQueue = new LinkedTransferQueue<>(); + private final Queue<OptimizingTableProcess> tableQueue = new LinkedTransferQueue<>(); private final SchedulingPolicy scheduler; private final CatalogManager catalogManager; private final Executor planExecutor; @@ -130,7 +114,7 @@ public class OptimizingQueue extends PersistentBase { private void initTableRuntime(DefaultTableRuntime tableRuntime) { try { - TableOptimizingProcess process = loadProcess(tableRuntime); + OptimizingTableProcess process = loadProcess(tableRuntime); if (!tableRuntime.getOptimizingConfig().isEnabled()) { closeProcessIfRunning(process); @@ -163,7 +147,7 @@ public class OptimizingQueue extends PersistentBase { } } - private TableOptimizingProcess loadProcess(DefaultTableRuntime tableRuntime) { + private OptimizingTableProcess loadProcess(DefaultTableRuntime tableRuntime) { if (tableRuntime.getProcessId() == 0) { return null; } @@ -184,11 +168,12 @@ public class OptimizingQueue extends PersistentBase { tableRuntime.getTableIdentifier()); return null; } - return new TableOptimizingProcess(tableRuntime, meta, state); + return new OptimizingTableProcess( + tableRuntime, meta, state, catalogManager, quotaProvider, optimizingTasksMap, null); } private boolean canResumeProcess( - TableOptimizingProcess process, DefaultTableRuntime tableRuntime) { + OptimizingTableProcess process, DefaultTableRuntime tableRuntime) { return process != null && process.getStatus() == ProcessStatus.RUNNING && tableRuntime.getOptimizingStatus().isProcessing() @@ -196,7 +181,7 @@ public class OptimizingQueue extends PersistentBase { } private void resetTableForRecovery( - TableOptimizingProcess process, DefaultTableRuntime tableRuntime) { + OptimizingTableProcess process, DefaultTableRuntime tableRuntime) { closeProcessIfRunning(process); if (tableRuntime.getOptimizingStatus() != OptimizingStatus.IDLE && tableRuntime.getOptimizingStatus() != OptimizingStatus.PENDING) { @@ -209,7 +194,7 @@ public class OptimizingQueue extends PersistentBase { } } - private void closeProcessIfRunning(TableOptimizingProcess process) { + private void closeProcessIfRunning(OptimizingTableProcess process) { if (process != null && process.getStatus() == ProcessStatus.RUNNING) { process.close(false); } @@ -373,7 +358,7 @@ public class OptimizingQueue extends PersistentBase { }); } - private TableOptimizingProcess planInternal(DefaultTableRuntime tableRuntime) { + private OptimizingTableProcess planInternal(DefaultTableRuntime tableRuntime) { tableRuntime.beginPlanning(); try { ServerTableIdentifier identifier = tableRuntime.getTableIdentifier(); @@ -385,7 +370,13 @@ public class OptimizingQueue extends PersistentBase { getAvailableCore(), maxInputSizePerThread()); if (planner.isNecessary()) { - return new TableOptimizingProcess(planner, tableRuntime); + return new OptimizingTableProcess( + planner, + tableRuntime, + catalogManager, + quotaProvider, + optimizingTasksMap, + () -> clearProcessById(planner.getProcessId())); } else { tableRuntime.completeEmptyProcess(); return null; @@ -397,6 +388,10 @@ public class OptimizingQueue extends PersistentBase { } } + private void clearProcessById(long processId) { + tableQueue.removeIf(process -> process.getProcessId() == processId); + } + public void ackTask(OptimizingTaskId taskId, OptimizerThread thread) { findProcess(taskId).ackTask(taskId, thread); } @@ -466,7 +461,7 @@ public class OptimizingQueue extends PersistentBase { this.metrics.unregister(); } - private TableOptimizingProcess findProcess(OptimizingTaskId taskId) { + private OptimizingTableProcess findProcess(OptimizingTaskId taskId) { return tableQueue.stream() .filter(p -> p.getProcessId() == taskId.getProcessId()) .findFirst() @@ -489,532 +484,4 @@ public class OptimizingQueue extends PersistentBase { SchedulingPolicy getSchedulingPolicy() { return scheduler; } - - private class TableOptimizingProcess implements OptimizingProcess { - - private final Lock lock = new ReentrantLock(); - private final long processId; - private final OptimizingType optimizingType; - private final DefaultTableRuntime tableRuntime; - private final long planTime; - private final long targetSnapshotId; - private final long targetChangeSnapshotId; - private final Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> taskMap = Maps.newHashMap(); - private final Queue<TaskRuntime<RewriteStageTask>> taskQueue = new LinkedList<>(); - private volatile ProcessStatus status = ProcessStatus.RUNNING; - private volatile String failedReason; - private long endTime = AmoroServiceConstants.INVALID_TIME; - private Map<String, Long> fromSequence = Maps.newHashMap(); - private Map<String, Long> toSequence = Maps.newHashMap(); - private boolean hasCommitted = false; - - public TaskRuntime<?> poll(OptimizerThread thread, boolean needQuotaChecking) { - try { - // Wait 10ms here for some light operation like poll/ack - if (lock.tryLock(10, TimeUnit.MILLISECONDS)) { - try { - TaskRuntime<?> task = null; - if (status != ProcessStatus.KILLED && status != ProcessStatus.FAILED) { - int actualQuota = getActualQuota(); - int quotaLimit = getQuotaLimit(); - if (!needQuotaChecking || actualQuota < quotaLimit) { - task = taskQueue.poll(); - } - } - if (task != null) { - optimizingTasksMap - .computeIfAbsent(tableRuntime.getTableIdentifier(), k -> new AtomicInteger(0)) - .incrementAndGet(); - task.schedule(thread); - } - return task; - } finally { - lock.unlock(); - } - } - } catch (InterruptedException e) { - // ignore it. - } - return null; - } - - public TableOptimizingProcess( - AbstractOptimizingPlanner planner, DefaultTableRuntime tableRuntime) { - processId = planner.getProcessId(); - this.tableRuntime = tableRuntime; - optimizingType = planner.getOptimizingType(); - planTime = planner.getPlanTime(); - targetSnapshotId = planner.getTargetSnapshotId(); - targetChangeSnapshotId = planner.getTargetChangeSnapshotId(); - loadTaskRuntimes(planner.planTasks()); - fromSequence = planner.getFromSequence(); - toSequence = planner.getToSequence(); - beginAndPersistProcess(); - } - - public TableOptimizingProcess( - DefaultTableRuntime tableRuntime, - TableProcessMeta processMeta, - OptimizingProcessState processState) { - this.tableRuntime = tableRuntime; - processId = tableRuntime.getProcessId(); - optimizingType = OptimizingType.valueOf(processMeta.getProcessType()); - targetSnapshotId = processState.getTargetSnapshotId(); - targetChangeSnapshotId = processState.getTargetChangeSnapshotId(); - planTime = processMeta.getCreateTime(); - if (processState.getFromSequence() != null) { - fromSequence = processState.getFromSequence(); - } - if (processState.getToSequence() != null) { - toSequence = processState.getToSequence(); - } - this.status = processMeta.getStatus(); - tableRuntime.recover(this); - loadTaskRuntimes(this); - } - - private int getQuotaLimit() { - double targetQuota = tableRuntime.getOptimizingConfig().getTargetQuota(); - return targetQuota > 1 - ? (int) targetQuota - : (int) Math.ceil(targetQuota * getAvailableCore()); - } - - @Override - public long getTableId() { - return tableRuntime.getTableIdentifier().getId(); - } - - @Override - public long getProcessId() { - return processId; - } - - @Override - public OptimizingType getOptimizingType() { - return optimizingType; - } - - @Override - public ProcessStatus getStatus() { - return status; - } - - @Override - public void close(boolean needCommit) { - lock.lock(); - try { - if (this.status != ProcessStatus.RUNNING) { - return; - } - if (tableRuntime.isAllowPartialCommit() && needCommit) { - tableRuntime.beginCommitting(); - } else { - this.status = ProcessStatus.CLOSED; - this.endTime = System.currentTimeMillis(); - persistAndSetCompleted(false); - } - } finally { - lock.unlock(); - } - } - - private void ackTask(OptimizingTaskId taskId, OptimizerThread thread) { - TaskRuntime<?> taskRuntime = getTaskRuntime(taskId); - lock.lock(); - try { - taskRuntime.ack(thread); - } finally { - lock.unlock(); - } - } - - private void completeTask(OptimizerThread thread, OptimizingTaskResult result) { - TaskRuntime<?> taskRuntime = getTaskRuntime(result.getTaskId()); - lock.lock(); - try { - taskRuntime.complete(thread, result); - } finally { - lock.unlock(); - } - } - - private TaskRuntime<?> getTaskRuntime(OptimizingTaskId taskId) { - TaskRuntime<?> taskRuntime = getTaskMap().get(taskId); - if (taskRuntime == null) { - throw new TaskNotFoundException(taskId); - } - return taskRuntime; - } - - private void acceptResult(TaskRuntime<?> taskRuntime) { - lock.lock(); - try { - optimizingTasksMap.computeIfPresent( - tableRuntime.getTableIdentifier(), - (k, v) -> { - if (v.get() > 0) { - v.decrementAndGet(); - } - return v; - }); - try { - tableRuntime.addTaskQuota(taskRuntime.getCurrentQuota()); - } catch (Throwable throwable) { - LOG.warn( - "{} failed to add task quota {}, ignore it", - tableRuntime.getTableIdentifier(), - taskRuntime.getTaskId(), - throwable); - } - // task cancel means persistAndSetCompleted has been called - if (taskRuntime.getStatus() == TaskRuntime.Status.CANCELED) { - return; - } - if (isClosed()) { - throw new OptimizingClosedException(processId); - } - if (taskRuntime.getStatus() == TaskRuntime.Status.SUCCESS) { - // the lock of TableOptimizingProcess makes it thread-safe - if (allTasksPrepared() - && tableRuntime.getOptimizingStatus().isProcessing() - && tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) { - tableRuntime.beginCommitting(); - } - } else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) { - if (taskRuntime.getRetry() - < tableRuntime.getOptimizingConfig().getMaxExecuteRetryCount()) { - LOG.info( - "Put task {} to retry queue, because {}", - taskRuntime.getTaskId(), - taskRuntime.getFailReason()); - retryTask(taskRuntime); - } else { - if (tableRuntime.isAllowPartialCommit() - && tableRuntime.getOptimizingStatus().isProcessing() - && tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) { - LOG.info( - "Task {} has reached the max execute retry count. Process {} cancels unfinished tasks and commits SUCCESS tasks.", - taskRuntime.getTaskId(), - processId); - failedReason = taskRuntime.getFailReason(); - tableRuntime.beginCommitting(); - } else { - LOG.info( - "Task {} has reached the max execute retry count. Process {} failed.", - taskRuntime.getTaskId(), - processId); - this.failedReason = taskRuntime.getFailReason(); - this.status = ProcessStatus.FAILED; - this.endTime = taskRuntime.getEndTime(); - persistAndSetCompleted(false); - } - } - } - } finally { - lock.unlock(); - } - } - - private void resetTask(TaskRuntime<RewriteStageTask> taskRuntime) { - lock.lock(); - try { - taskRuntime.reset(); - taskQueue.add(taskRuntime); - } finally { - lock.unlock(); - } - } - - @Override - public boolean isClosed() { - return status == ProcessStatus.KILLED; - } - - @Override - public long getPlanTime() { - return planTime; - } - - @Override - public long getDuration() { - long dur = - endTime == AmoroServiceConstants.INVALID_TIME - ? System.currentTimeMillis() - planTime - : endTime - planTime; - return Math.max(0, dur); - } - - @Override - public long getTargetSnapshotId() { - return targetSnapshotId; - } - - @Override - public long getTargetChangeSnapshotId() { - return targetChangeSnapshotId; - } - - public String getFailedReason() { - return failedReason; - } - - private Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> getTaskMap() { - return taskMap; - } - - /** - * if all tasks are Prepared - * - * @return true if tasks is not empty and all Prepared - */ - private boolean allTasksPrepared() { - if (!taskMap.isEmpty()) { - return taskMap.values().stream().allMatch(t -> t.getStatus() == TaskRuntime.Status.SUCCESS); - } - return false; - } - - /** - * Get optimizeRuntime. - * - * @return - - */ - @Override - public long getRunningQuotaTime(long calculatingStartTime, long calculatingEndTime) { - return taskMap.values().stream() - .filter(t -> !t.finished()) - .mapToLong(task -> task.getQuotaTime(calculatingStartTime, calculatingEndTime)) - .sum(); - } - - public int getActualQuota() { - return optimizingTasksMap - .getOrDefault(tableRuntime.getTableIdentifier(), new AtomicInteger(0)) - .get(); - } - - @Override - public void commit() { - List<TaskRuntime<RewriteStageTask>> successTasks = - taskMap.values().stream() - .filter(task -> task.getStatus() == Status.SUCCESS) - .collect(Collectors.toList()); - LOG.debug( - "{} get {} tasks of {} partitions to commit", - tableRuntime.getTableIdentifier(), - successTasks.size(), - successTasks.stream() - .map(task -> task.getTaskDescriptor().getPartition()) - .distinct() - .count()); - - lock.lock(); - try { - if (hasCommitted) { - LOG.warn("{} has already committed, give up", tableRuntime.getTableIdentifier()); - try { - persistAndSetCompleted(status == ProcessStatus.SUCCESS); - } catch (Exception ignored) { - } - throw new IllegalStateException("repeat commit, and last error " + failedReason); - } - try { - hasCommitted = true; - buildCommit().commit(); - if (allTasksPrepared()) { - status = ProcessStatus.SUCCESS; - } else if (taskMap.values().stream() - .anyMatch(task -> task.getStatus() == TaskRuntime.Status.FAILED)) { - status = ProcessStatus.FAILED; - } else { - status = ProcessStatus.CLOSED; - } - endTime = System.currentTimeMillis(); - persistAndSetCompleted(status == ProcessStatus.SUCCESS); - } catch (PersistenceException e) { - LOG.warn( - "{} failed to persist process completed, will retry next commit", - tableRuntime.getTableIdentifier(), - e); - } catch (Throwable t) { - LOG.error("{} Commit optimizing failed ", tableRuntime.getTableIdentifier(), t); - status = ProcessStatus.FAILED; - failedReason = ExceptionUtil.getErrorMessage(t, 4000); - endTime = System.currentTimeMillis(); - persistAndSetCompleted(false); - } - } finally { - lock.unlock(); - } - } - - @Override - public MetricsSummary getSummary() { - List<MetricsSummary> taskSummaries = - taskMap.values().stream() - .map(TaskRuntime::getTaskDescriptor) - .map(RewriteStageTask::getSummary) - .collect(Collectors.toList()); - - return new MetricsSummary(taskSummaries); - } - - private UnKeyedTableCommit buildCommit() { - MixedTable table = - (MixedTable) - catalogManager - .loadTable(tableRuntime.getTableIdentifier().getIdentifier()) - .originalTable(); - if (table.isUnkeyedTable()) { - return new UnKeyedTableCommit(targetSnapshotId, table, taskMap.values()); - } else { - return new KeyedTableCommit( - table, - taskMap.values(), - targetSnapshotId, - convertPartitionSequence(table, fromSequence), - convertPartitionSequence(table, toSequence)); - } - } - - private StructLikeMap<Long> convertPartitionSequence( - MixedTable table, Map<String, Long> partitionSequence) { - PartitionSpec spec = table.spec(); - StructLikeMap<Long> results = StructLikeMap.create(spec.partitionType()); - partitionSequence.forEach( - (partition, sequence) -> { - if (spec.isUnpartitioned()) { - results.put(TablePropertyUtil.EMPTY_STRUCT, sequence); - } else { - StructLike partitionData = MixedDataFiles.data(spec, partition); - results.put(partitionData, sequence); - } - }); - return results; - } - - private void beginAndPersistProcess() { - doAsTransaction( - () -> - doAs( - TableProcessMapper.class, - mapper -> - mapper.insertProcess( - tableRuntime.getTableIdentifier().getId(), - processId, - "", - status, - optimizingType.name().toUpperCase(), - tableRuntime.getOptimizingStatus().name().toLowerCase(), - "AMORO", - 0, - planTime, - new HashMap<>(), - getSummary().summaryAsMap(false))), - () -> - doAs( - OptimizingProcessMapper.class, - mapper -> - mapper.insertInternalProcessState( - tableRuntime.getTableIdentifier().getId(), - processId, - targetSnapshotId, - targetChangeSnapshotId, - fromSequence, - toSequence)), - () -> - doAs( - OptimizingProcessMapper.class, - mapper -> mapper.insertTaskRuntimes(Lists.newArrayList(taskMap.values()))), - () -> TaskFilesPersistence.persistTaskInputs(processId, taskMap.values()), - () -> tableRuntime.beginProcess(this)); - } - - private void persistAndSetCompleted(boolean success) { - doAsTransaction( - () -> { - if (!success) { - cancelTasks(); - } - }, - () -> - doAs( - TableProcessMapper.class, - mapper -> - mapper.updateProcess( - tableRuntime.getTableIdentifier().getId(), - processId, - "", - status, - tableRuntime.getOptimizingStatus().name().toLowerCase(), - 0, - System.currentTimeMillis(), - getFailedReason(), - new HashMap<>(), - getSummary().summaryAsMap(false))), - () -> tableRuntime.completeProcess(success), - () -> clearProcess(this)); - } - - private void cancelTasks() { - taskMap.values().forEach(TaskRuntime::tryCanceling); - } - - private void loadTaskRuntimes(OptimizingProcess optimizingProcess) { - try { - List<TaskRuntime<RewriteStageTask>> taskRuntimes = - getAs( - OptimizingProcessMapper.class, - mapper -> - mapper.selectTaskRuntimes( - tableRuntime.getTableIdentifier().getId(), processId)); - Map<Integer, RewriteFilesInput> inputs = TaskFilesPersistence.loadTaskInputs(processId); - taskRuntimes.forEach( - taskRuntime -> { - taskRuntime.getCompletedFuture().whenCompleted(() -> acceptResult(taskRuntime)); - taskRuntime - .getTaskDescriptor() - .setInput(inputs.get(taskRuntime.getTaskId().getTaskId())); - taskMap.put(taskRuntime.getTaskId(), taskRuntime); - if (taskRuntime.getStatus() == TaskRuntime.Status.PLANNED) { - taskQueue.offer(taskRuntime); - } else if (taskRuntime.getStatus() == TaskRuntime.Status.SCHEDULED - || taskRuntime.getStatus() == TaskRuntime.Status.ACKED) { - // Don't reset — let the optimizer finish execution and report - // the result. If the optimizer is dead, OptimizerKeeper will - // detect the stale token and reset the task via retryTask(). - LOG.debug( - "Keeping task {} in {} during recovery, waiting for optimizer to complete", - taskRuntime.getTaskId(), - taskRuntime.getStatus()); - } else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) { - retryTask(taskRuntime); - } - }); - } catch (IllegalArgumentException e) { - LOG.warn( - "Load task inputs failed, close the optimizing process : {}", - optimizingProcess.getProcessId(), - e); - optimizingProcess.close(false); - } - } - - private void loadTaskRuntimes(List<RewriteStageTask> taskDescriptors) { - int taskId = 1; - for (RewriteStageTask taskDescriptor : taskDescriptors) { - TaskRuntime<RewriteStageTask> taskRuntime = - new TaskRuntime<>(new OptimizingTaskId(processId, taskId++), taskDescriptor); - LOG.info( - "{} plan new task {}, summary {}", - tableRuntime.getTableIdentifier(), - taskRuntime.getTaskId(), - taskRuntime.getSummary()); - 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/OptimizingTableProcess.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingTableProcess.java new file mode 100644 index 000000000..6523f0535 --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingTableProcess.java @@ -0,0 +1,630 @@ +/* + * 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; + +import org.apache.amoro.ServerTableIdentifier; +import org.apache.amoro.api.OptimizingTaskId; +import org.apache.amoro.api.OptimizingTaskResult; +import org.apache.amoro.exception.OptimizingClosedException; +import org.apache.amoro.exception.PersistenceException; +import org.apache.amoro.exception.TaskNotFoundException; +import org.apache.amoro.optimizing.MetricsSummary; +import org.apache.amoro.optimizing.OptimizingType; +import org.apache.amoro.optimizing.RewriteFilesInput; +import org.apache.amoro.optimizing.RewriteStageTask; +import org.apache.amoro.optimizing.TableOptimizingCommitter; +import org.apache.amoro.optimizing.plan.AbstractOptimizingPlanner; +import org.apache.amoro.process.ProcessStatus; +import org.apache.amoro.server.AmoroServiceConstants; +import org.apache.amoro.server.catalog.CatalogManager; +import org.apache.amoro.server.optimizing.TaskRuntime.Status; +import org.apache.amoro.server.persistence.OptimizingProcessState; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TaskFilesPersistence; +import org.apache.amoro.server.persistence.mapper.OptimizingProcessMapper; +import org.apache.amoro.server.persistence.mapper.TableProcessMapper; +import org.apache.amoro.server.process.TableProcessMeta; +import org.apache.amoro.server.resource.OptimizerThread; +import org.apache.amoro.server.resource.QuotaProvider; +import org.apache.amoro.server.table.DefaultTableRuntime; +import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.table.MixedTable; +import org.apache.amoro.utils.ExceptionUtil; +import org.apache.amoro.utils.MixedDataFiles; +import org.apache.amoro.utils.TablePropertyUtil; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.util.StructLikeMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +/** + * Standalone class for managing the lifecycle of an optimizing process for a table. Extracted from + * OptimizingQueue.TableOptimizingProcess inner class. + */ +public class OptimizingTableProcess extends PersistentBase implements OptimizingProcess { + + private static final Logger LOG = LoggerFactory.getLogger(OptimizingTableProcess.class); + + private final Lock lock = new ReentrantLock(); + private final long processId; + private final OptimizingType optimizingType; + private final DefaultTableRuntime tableRuntime; + private final long planTime; + private final long targetSnapshotId; + private final long targetChangeSnapshotId; + private final Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> taskMap = Maps.newHashMap(); + private final Queue<TaskRuntime<RewriteStageTask>> taskQueue = new LinkedList<>(); + private volatile ProcessStatus status = ProcessStatus.RUNNING; + private volatile String failedReason; + private long endTime = AmoroServiceConstants.INVALID_TIME; + private Map<String, Long> fromSequence = Maps.newHashMap(); + private Map<String, Long> toSequence = Maps.newHashMap(); + private boolean hasCommitted = false; + + // Injected dependencies (previously accessed from enclosing OptimizingQueue) + private final CatalogManager catalogManager; + private final QuotaProvider quotaProvider; + private final Map<ServerTableIdentifier, AtomicInteger> optimizingTasksMap; + private final Runnable onProcessCompleted; + private IcebergOptimizingProcessFactory processFactory; + + /** Constructor for creating a new process from a planner result. */ + public OptimizingTableProcess( + AbstractOptimizingPlanner planner, + DefaultTableRuntime tableRuntime, + CatalogManager catalogManager, + QuotaProvider quotaProvider, + Map<ServerTableIdentifier, AtomicInteger> optimizingTasksMap, + Runnable onProcessCompleted) { + this.tableRuntime = tableRuntime; + this.catalogManager = catalogManager; + this.quotaProvider = quotaProvider; + this.optimizingTasksMap = optimizingTasksMap; + this.onProcessCompleted = onProcessCompleted; + this.processId = planner.getProcessId(); + this.optimizingType = planner.getOptimizingType(); + this.planTime = planner.getPlanTime(); + this.targetSnapshotId = planner.getTargetSnapshotId(); + this.targetChangeSnapshotId = planner.getTargetChangeSnapshotId(); + loadTaskRuntimes(planner.planTasks()); + this.fromSequence = planner.getFromSequence(); + this.toSequence = planner.getToSequence(); + beginAndPersistProcess(); + } + + /** Constructor for recovering a process from persisted state. */ + public OptimizingTableProcess( + DefaultTableRuntime tableRuntime, + TableProcessMeta processMeta, + OptimizingProcessState processState, + CatalogManager catalogManager, + QuotaProvider quotaProvider, + Map<ServerTableIdentifier, AtomicInteger> optimizingTasksMap, + Runnable onProcessCompleted) { + this.tableRuntime = tableRuntime; + this.catalogManager = catalogManager; + this.quotaProvider = quotaProvider; + this.optimizingTasksMap = optimizingTasksMap; + this.onProcessCompleted = onProcessCompleted; + this.processId = tableRuntime.getProcessId(); + this.optimizingType = OptimizingType.valueOf(processMeta.getProcessType()); + this.targetSnapshotId = processState.getTargetSnapshotId(); + this.targetChangeSnapshotId = processState.getTargetChangeSnapshotId(); + this.planTime = processMeta.getCreateTime(); + if (processState.getFromSequence() != null) { + this.fromSequence = processState.getFromSequence(); + } + if (processState.getToSequence() != null) { + this.toSequence = processState.getToSequence(); + } + this.status = processMeta.getStatus(); + tableRuntime.recover(this); + loadTaskRuntimes(this); + } + + public TaskRuntime<?> poll(OptimizerThread thread, boolean needQuotaChecking) { + try { + if (lock.tryLock(10, TimeUnit.MILLISECONDS)) { + try { + TaskRuntime<?> task = null; + if (status != ProcessStatus.KILLED && status != ProcessStatus.FAILED) { + int actualQuota = getActualQuota(); + int quotaLimit = getQuotaLimit(); + if (!needQuotaChecking || actualQuota < quotaLimit) { + task = taskQueue.poll(); + } + } + if (task != null) { + optimizingTasksMap + .computeIfAbsent(tableRuntime.getTableIdentifier(), k -> new AtomicInteger(0)) + .incrementAndGet(); + task.schedule(thread); + } + return task; + } finally { + lock.unlock(); + } + } + } catch (InterruptedException e) { + // ignore it. + } + return null; + } + + public void ackTask(OptimizingTaskId taskId, OptimizerThread thread) { + TaskRuntime<?> taskRuntime = getTaskRuntime(taskId); + lock.lock(); + try { + taskRuntime.ack(thread); + } finally { + lock.unlock(); + } + } + + public void completeTask(OptimizerThread thread, OptimizingTaskResult result) { + TaskRuntime<?> taskRuntime = getTaskRuntime(result.getTaskId()); + lock.lock(); + try { + taskRuntime.complete(thread, result); + } finally { + lock.unlock(); + } + } + + public void resetTask(TaskRuntime<RewriteStageTask> taskRuntime) { + lock.lock(); + try { + taskRuntime.reset(); + taskQueue.add(taskRuntime); + } finally { + lock.unlock(); + } + } + + private TaskRuntime<?> getTaskRuntime(OptimizingTaskId taskId) { + TaskRuntime<?> taskRuntime = getTaskMap().get(taskId); + if (taskRuntime == null) { + throw new TaskNotFoundException(taskId); + } + return taskRuntime; + } + + private void acceptResult(TaskRuntime<?> taskRuntime) { + lock.lock(); + try { + optimizingTasksMap.computeIfPresent( + tableRuntime.getTableIdentifier(), + (k, v) -> { + if (v.get() > 0) { + v.decrementAndGet(); + } + return v; + }); + try { + tableRuntime.addTaskQuota(taskRuntime.getCurrentQuota()); + } catch (Throwable throwable) { + LOG.warn( + "{} failed to add task quota {}, ignore it", + tableRuntime.getTableIdentifier(), + taskRuntime.getTaskId(), + throwable); + } + if (taskRuntime.getStatus() == TaskRuntime.Status.CANCELED) { + return; + } + if (isClosed()) { + throw new OptimizingClosedException(processId); + } + if (taskRuntime.getStatus() == TaskRuntime.Status.SUCCESS) { + if (allTasksPrepared() + && tableRuntime.getOptimizingStatus().isProcessing() + && tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) { + tableRuntime.beginCommitting(); + } + } else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) { + if (taskRuntime.getRetry() < tableRuntime.getOptimizingConfig().getMaxExecuteRetryCount()) { + LOG.info( + "Put task {} to retry queue, because {}", + taskRuntime.getTaskId(), + taskRuntime.getFailReason()); + resetTask((TaskRuntime<RewriteStageTask>) taskRuntime); + } else { + if (tableRuntime.isAllowPartialCommit() + && tableRuntime.getOptimizingStatus().isProcessing() + && tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) { + LOG.info( + "Task {} has reached the max execute retry count. Process {} cancels " + + "unfinished tasks and commits SUCCESS tasks.", + taskRuntime.getTaskId(), + processId); + failedReason = taskRuntime.getFailReason(); + tableRuntime.beginCommitting(); + } else { + LOG.info( + "Task {} has reached the max execute retry count. Process {} failed.", + taskRuntime.getTaskId(), + processId); + this.failedReason = taskRuntime.getFailReason(); + this.status = ProcessStatus.FAILED; + this.endTime = taskRuntime.getEndTime(); + persistAndSetCompleted(false); + } + } + } + } finally { + lock.unlock(); + } + } + + private int getQuotaLimit() { + double targetQuota = tableRuntime.getOptimizingConfig().getTargetQuota(); + return targetQuota > 1 ? (int) targetQuota : (int) Math.ceil(targetQuota * getAvailableCore()); + } + + private double getAvailableCore() { + return Math.max(quotaProvider.getTotalQuota(tableRuntime.getGroupName()), 1); + } + + // ===== OptimizingProcess interface implementation ===== + + @Override + public long getTableId() { + return tableRuntime.getTableIdentifier().getId(); + } + + @Override + public long getProcessId() { + return processId; + } + + @Override + public OptimizingType getOptimizingType() { + return optimizingType; + } + + @Override + public ProcessStatus getStatus() { + return status; + } + + @Override + public void close(boolean needCommit) { + lock.lock(); + try { + if (this.status != ProcessStatus.RUNNING) { + return; + } + if (tableRuntime.isAllowPartialCommit() && needCommit) { + tableRuntime.beginCommitting(); + } else { + this.status = ProcessStatus.CLOSED; + this.endTime = System.currentTimeMillis(); + persistAndSetCompleted(false); + } + } finally { + lock.unlock(); + } + } + + @Override + public boolean isClosed() { + return status == ProcessStatus.KILLED; + } + + @Override + public long getPlanTime() { + return planTime; + } + + @Override + public long getDuration() { + long dur = + endTime == AmoroServiceConstants.INVALID_TIME + ? System.currentTimeMillis() - planTime + : endTime - planTime; + return Math.max(0, dur); + } + + @Override + public long getTargetSnapshotId() { + return targetSnapshotId; + } + + @Override + public long getTargetChangeSnapshotId() { + return targetChangeSnapshotId; + } + + public String getFailedReason() { + return failedReason; + } + + public void setProcessFactory(IcebergOptimizingProcessFactory processFactory) { + this.processFactory = processFactory; + } + + public Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> getTaskMap() { + return taskMap; + } + + public boolean allTasksPrepared() { + if (!taskMap.isEmpty()) { + return taskMap.values().stream().allMatch(t -> t.getStatus() == TaskRuntime.Status.SUCCESS); + } + return false; + } + + @Override + public long getRunningQuotaTime(long calculatingStartTime, long calculatingEndTime) { + return taskMap.values().stream() + .filter(t -> !t.finished()) + .mapToLong(task -> task.getQuotaTime(calculatingStartTime, calculatingEndTime)) + .sum(); + } + + public int getActualQuota() { + return optimizingTasksMap + .getOrDefault(tableRuntime.getTableIdentifier(), new AtomicInteger(0)) + .get(); + } + + @Override + public void commit() { + List<TaskRuntime<RewriteStageTask>> successTasks = + taskMap.values().stream() + .filter(task -> task.getStatus() == Status.SUCCESS) + .collect(Collectors.toList()); + LOG.debug( + "{} get {} tasks of {} partitions to commit", + tableRuntime.getTableIdentifier(), + successTasks.size(), + successTasks.stream() + .map(task -> task.getTaskDescriptor().getPartition()) + .distinct() + .count()); + + lock.lock(); + try { + if (hasCommitted) { + LOG.warn("{} has already committed, give up", tableRuntime.getTableIdentifier()); + try { + persistAndSetCompleted(status == ProcessStatus.SUCCESS); + } catch (Exception ignored) { + } + throw new IllegalStateException("repeat commit, and last error " + failedReason); + } + try { + hasCommitted = true; + buildCommit().commit(); + if (allTasksPrepared()) { + status = ProcessStatus.SUCCESS; + } else if (taskMap.values().stream() + .anyMatch(task -> task.getStatus() == TaskRuntime.Status.FAILED)) { + status = ProcessStatus.FAILED; + } else { + status = ProcessStatus.CLOSED; + } + endTime = System.currentTimeMillis(); + persistAndSetCompleted(status == ProcessStatus.SUCCESS); + } catch (PersistenceException e) { + LOG.warn( + "{} failed to persist process completed, will retry next commit", + tableRuntime.getTableIdentifier(), + e); + } catch (Throwable t) { + LOG.error("{} Commit optimizing failed ", tableRuntime.getTableIdentifier(), t); + status = ProcessStatus.FAILED; + failedReason = ExceptionUtil.getErrorMessage(t, 4000); + endTime = System.currentTimeMillis(); + persistAndSetCompleted(false); + } + } finally { + lock.unlock(); + } + } + + @Override + public MetricsSummary getSummary() { + List<MetricsSummary> taskSummaries = + taskMap.values().stream() + .map(TaskRuntime::getTaskDescriptor) + .map(RewriteStageTask::getSummary) + .collect(Collectors.toList()); + + return new MetricsSummary(taskSummaries); + } + + // ===== Private methods ===== + + private TableOptimizingCommitter buildCommit() { + MixedTable table = + (MixedTable) + catalogManager + .loadTable(tableRuntime.getTableIdentifier().getIdentifier()) + .originalTable(); + // Use factory callback for commit if available, otherwise fallback to direct creation + if (processFactory != null) { + return processFactory.createCommitter( + table, targetSnapshotId, taskMap.values(), fromSequence, toSequence); + } + // Fallback: direct creation (for backwards compatibility during migration) + if (table.isUnkeyedTable()) { + return new UnKeyedTableCommit(targetSnapshotId, table, taskMap.values()); + } else { + return new KeyedTableCommit( + table, + taskMap.values(), + targetSnapshotId, + convertPartitionSequence(table, fromSequence), + convertPartitionSequence(table, toSequence)); + } + } + + private StructLikeMap<Long> convertPartitionSequence( + MixedTable table, Map<String, Long> partitionSequence) { + PartitionSpec spec = table.spec(); + StructLikeMap<Long> results = StructLikeMap.create(spec.partitionType()); + partitionSequence.forEach( + (partition, sequence) -> { + if (spec.isUnpartitioned()) { + results.put(TablePropertyUtil.EMPTY_STRUCT, sequence); + } else { + StructLike partitionData = MixedDataFiles.data(spec, partition); + results.put(partitionData, sequence); + } + }); + return results; + } + + private void beginAndPersistProcess() { + doAsTransaction( + () -> + doAs( + TableProcessMapper.class, + mapper -> + mapper.insertProcess( + tableRuntime.getTableIdentifier().getId(), + processId, + "", + status, + optimizingType.name().toUpperCase(), + tableRuntime.getOptimizingStatus().name().toLowerCase(), + "AMORO", + 0, + planTime, + new HashMap<>(), + getSummary().summaryAsMap(false))), + () -> + doAs( + OptimizingProcessMapper.class, + mapper -> + mapper.insertInternalProcessState( + tableRuntime.getTableIdentifier().getId(), + processId, + targetSnapshotId, + targetChangeSnapshotId, + fromSequence, + toSequence)), + () -> + doAs( + OptimizingProcessMapper.class, + mapper -> mapper.insertTaskRuntimes(Lists.newArrayList(taskMap.values()))), + () -> TaskFilesPersistence.persistTaskInputs(processId, taskMap.values()), + () -> tableRuntime.beginProcess(this)); + } + + private void persistAndSetCompleted(boolean success) { + doAsTransaction( + () -> { + if (!success) { + cancelTasks(); + } + }, + () -> + doAs( + TableProcessMapper.class, + mapper -> + mapper.updateProcess( + tableRuntime.getTableIdentifier().getId(), + processId, + "", + status, + tableRuntime.getOptimizingStatus().name().toLowerCase(), + 0, + System.currentTimeMillis(), + getFailedReason(), + new HashMap<>(), + getSummary().summaryAsMap(false))), + () -> tableRuntime.completeProcess(success), + () -> { + if (onProcessCompleted != null) { + onProcessCompleted.run(); + } + }); + } + + private void cancelTasks() { + taskMap.values().forEach(TaskRuntime::tryCanceling); + } + + private void loadTaskRuntimes(OptimizingProcess optimizingProcess) { + try { + List<TaskRuntime<RewriteStageTask>> taskRuntimes = + getAs( + OptimizingProcessMapper.class, + mapper -> + mapper.selectTaskRuntimes(tableRuntime.getTableIdentifier().getId(), processId)); + Map<Integer, RewriteFilesInput> inputs = TaskFilesPersistence.loadTaskInputs(processId); + taskRuntimes.forEach( + taskRuntime -> { + taskRuntime.getCompletedFuture().whenCompleted(() -> acceptResult(taskRuntime)); + taskRuntime + .getTaskDescriptor() + .setInput(inputs.get(taskRuntime.getTaskId().getTaskId())); + taskMap.put(taskRuntime.getTaskId(), taskRuntime); + if (taskRuntime.getStatus() == TaskRuntime.Status.PLANNED) { + taskQueue.offer(taskRuntime); + } else if (taskRuntime.getStatus() == TaskRuntime.Status.SCHEDULED + || taskRuntime.getStatus() == TaskRuntime.Status.ACKED) { + LOG.debug( + "Keeping task {} in {} during recovery, waiting for optimizer to complete", + taskRuntime.getTaskId(), + taskRuntime.getStatus()); + } else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) { + resetTask(taskRuntime); + } + }); + } catch (IllegalArgumentException e) { + LOG.warn( + "Load task inputs failed, close the optimizing process : {}", + optimizingProcess.getProcessId(), + e); + optimizingProcess.close(false); + } + } + + private void loadTaskRuntimes(List<RewriteStageTask> taskDescriptors) { + int taskId = 1; + for (RewriteStageTask taskDescriptor : taskDescriptors) { + TaskRuntime<RewriteStageTask> taskRuntime = + new TaskRuntime<>(new OptimizingTaskId(processId, taskId++), taskDescriptor); + LOG.info( + "{} plan new task {}, summary {}", + tableRuntime.getTableIdentifier(), + taskRuntime.getTaskId(), + taskRuntime.getSummary()); + 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/UnKeyedTableCommit.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java index d71ae484b..19f07dbad 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java @@ -33,6 +33,7 @@ import org.apache.amoro.iceberg.Constants; import org.apache.amoro.op.SnapshotSummary; import org.apache.amoro.optimizing.RewriteFilesOutput; import org.apache.amoro.optimizing.RewriteStageTask; +import org.apache.amoro.optimizing.TableOptimizingCommitter; import org.apache.amoro.optimizing.TaskProperties; import org.apache.amoro.properties.HiveTableProperties; import org.apache.amoro.server.optimizing.TaskRuntime.Status; @@ -74,7 +75,7 @@ import java.util.Map; import java.util.Set; import java.util.stream.Collectors; -public class UnKeyedTableCommit { +public class UnKeyedTableCommit implements TableOptimizingCommitter { private static final Logger LOG = LoggerFactory.getLogger(UnKeyedTableCommit.class); private final Long targetSnapshotId; diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/ActionCoordinatorScheduler.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/ActionCoordinatorScheduler.java index 4e9f79a35..3257663b1 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/ActionCoordinatorScheduler.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/ActionCoordinatorScheduler.java @@ -18,21 +18,30 @@ package org.apache.amoro.server.process; +import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.TableFormat; import org.apache.amoro.TableRuntime; +import org.apache.amoro.api.BlockableOperation; import org.apache.amoro.process.ActionCoordinator; import org.apache.amoro.process.TableProcess; import org.apache.amoro.process.TableProcessStore; +import org.apache.amoro.server.persistence.SqlSessionFactoryProvider; +import org.apache.amoro.server.persistence.mapper.TableBlockerMapper; import org.apache.amoro.server.scheduler.PeriodicTableScheduler; import org.apache.amoro.server.table.TableService; +import org.apache.amoro.server.table.blocker.TableBlocker; +import org.apache.ibatis.session.SqlSession; +import org.apache.ibatis.session.TransactionIsolationLevel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.List; import java.util.Optional; /** * Periodic scheduler that delegates scheduling decisions to an {@link ActionCoordinator}. It - * creates, recovers and retries table processes via {@link ProcessService}. + * creates, recovers and retries table processes via {@link ProcessService}. Includes Blocker + * checking for OPTIMIZE operations. */ public class ActionCoordinatorScheduler extends PeriodicTableScheduler { @@ -49,65 +58,43 @@ public class ActionCoordinatorScheduler extends PeriodicTableScheduler { this.processService = processService; } - /** - * Get the bound coordinator. - * - * @return coordinator - */ + /** Get the bound coordinator. */ public ActionCoordinator getCoordinator() { return coordinator; } - /** - * Whether the given table format is supported. - * - * @param format table format - * @return true if supported - */ @Override protected boolean formatSupported(TableFormat format) { return coordinator.formatSupported(format); } - /** - * Compute next executing time for a table runtime. - * - * @param tableRuntime table runtime - * @return next executing timestamp in milliseconds - */ @Override protected long getNextExecutingTime(TableRuntime tableRuntime) { return coordinator.getNextExecutingTime(tableRuntime); } - /** - * Whether the given table runtime is enabled for scheduling. - * - * @param tableRuntime table runtime - * @return true if enabled - */ @Override protected boolean enabled(TableRuntime tableRuntime) { return coordinator.enabled(tableRuntime); } /** - * Create and register a new table process for the given table runtime. - * - * @param tableRuntime table runtime + * Create and register a new table process for the given table runtime. Checks for Blocker + * conflicts before triggering. */ @Override protected void execute(TableRuntime tableRuntime) { + // Check for Blocker conflicts (Step 6.6 from the refactoring plan) + if (isBlocked(tableRuntime)) { + LOG.debug( + "Table {} is blocked for optimize, skip scheduling", tableRuntime.getTableIdentifier()); + return; + } Optional<TableProcess> process = coordinator.trigger(tableRuntime); process.ifPresent(p -> processService.register(tableRuntime, p)); } - /** - * Recover and register a table process from store. - * - * @param tableRuntime table runtime - * @param processStore process store - */ + /** Recover and register a table process from store. */ protected TableProcess recover(TableRuntime tableRuntime, TableProcessStore processStore) { return coordinator.recoverTableProcess(tableRuntime, processStore); } @@ -116,4 +103,32 @@ public class ActionCoordinatorScheduler extends PeriodicTableScheduler { protected long getExecutorDelay() { return coordinator.getExecutorDelay(); } + + /** + * Check if the table is blocked for the OPTIMIZE operation. Migrated from + * OptimizingQueue.skipBlockedTables(). + */ + private boolean isBlocked(TableRuntime tableRuntime) { + try (SqlSession session = + SqlSessionFactoryProvider.getInstance() + .get() + .openSession(TransactionIsolationLevel.READ_COMMITTED)) { + List<TableBlocker> blockers = + session.getMapper(TableBlockerMapper.class).selectAllBlockers(System.currentTimeMillis()); + ServerTableIdentifier identifier = tableRuntime.getTableIdentifier(); + return blockers.stream() + .anyMatch( + blocker -> + TableBlocker.conflict(BlockableOperation.OPTIMIZE, blocker) + && blocker.getCatalog().equals(identifier.getCatalog()) + && blocker.getDatabase().equals(identifier.getDatabase()) + && blocker.getTableName().equals(identifier.getTableName())); + } catch (Exception e) { + LOG.warn( + "Failed to check blockers for table {}, proceeding", + tableRuntime.getTableIdentifier(), + e); + return false; + } + } } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/resource/OptimizerRegistryService.java b/amoro-ams/src/main/java/org/apache/amoro/server/resource/OptimizerRegistryService.java new file mode 100644 index 000000000..ab3e381ce --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/resource/OptimizerRegistryService.java @@ -0,0 +1,216 @@ +/* + * 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.resource; + +import org.apache.amoro.api.OptimizerRegisterInfo; +import org.apache.amoro.server.optimizing.TaskRuntime; +import org.apache.amoro.server.persistence.StatedPersistentBase; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; + +/** + * Manages optimizer registration, authentication, heartbeat, and timeout detection. Extracted from + * DefaultOptimizingService. + */ +public class OptimizerRegistryService extends StatedPersistentBase { + + private static final Logger LOG = LoggerFactory.getLogger(OptimizerRegistryService.class); + + private final long optimizerTouchTimeout; + private final long taskAckTimeout; + private final long taskExecuteTimeout; + private final QuotaManager quotaManager; + private final Map<String, OptimizerInstance> authOptimizers = new ConcurrentHashMap<>(); + private final OptimizerKeeper optimizerKeeper; + + public OptimizerRegistryService( + long optimizerTouchTimeout, + long taskAckTimeout, + long taskExecuteTimeout, + QuotaManager quotaManager) { + this.optimizerTouchTimeout = optimizerTouchTimeout; + this.taskAckTimeout = taskAckTimeout; + this.taskExecuteTimeout = taskExecuteTimeout; + this.quotaManager = quotaManager; + this.optimizerKeeper = new OptimizerKeeper(); + } + + public String authenticate(OptimizerRegisterInfo registerInfo) { + String token = UUID.randomUUID().toString(); + OptimizerInstance optimizer = new OptimizerInstance(registerInfo, token); + registerOptimizer(optimizer, true); + authOptimizers.put(token, optimizer); + LOG.info( + "Authenticated optimizer {} from group {} with token {}", + optimizer.getResourceId(), + optimizer.getGroupName(), + token); + return token; + } + + public void touch(String authToken) { + OptimizerInstance optimizer = authOptimizers.get(authToken); + if (optimizer == null) { + LOG.warn("Unknown auth token for touch: {}", authToken); + return; + } + optimizer.touch(); + } + + public void registerOptimizer(OptimizerInstance optimizer, boolean needPersistent) { + if (needPersistent) { + doAs(OptimizerMapper.class, mapper -> mapper.insertOptimizer(optimizer)); + } + quotaManager.addOptimizer(optimizer); + } + + public void unregisterOptimizer(String token) { + OptimizerInstance optimizer = authOptimizers.remove(token); + if (optimizer != null) { + quotaManager.removeOptimizer(optimizer); + LOG.info( + "Unregistered optimizer {} from group {}", + optimizer.getResourceId(), + optimizer.getGroupName()); + } + } + + public OptimizerInstance getAuthenticatedOptimizer(String authToken) { + return authOptimizers.get(authToken); + } + + public OptimizerThread getOptimizerThread(String authToken, int threadId) { + OptimizerInstance optimizer = authOptimizers.get(authToken); + if (optimizer == null) { + return null; + } + return optimizer.getThread(threadId); + } + + public String getGroupNameByToken(String authToken) { + OptimizerInstance optimizer = authOptimizers.get(authToken); + return optimizer != null ? optimizer.getGroupName() : null; + } + + public Set<String> getActiveTokens() { + return Collections.unmodifiableSet(authOptimizers.keySet()); + } + + public void startKeeper( + Predicate<TaskRuntime<?>> taskRetryPredicate, + java.util.function.Function<String, List<TaskRuntime<?>>> taskCollector, + java.util.function.Consumer<TaskRuntime<?>> taskRetryHandler) { + optimizerKeeper.start(taskRetryPredicate, taskCollector, taskRetryHandler); + } + + public void stopKeeper() { + optimizerKeeper.stop(); + } + + public void deleteOptimizer(String group, String resourceId) { + doAs(OptimizerMapper.class, mapper -> mapper.deleteOptimizer(resourceId)); + } + + /** Inner class for detecting expired optimizers and suspended tasks. */ + private class OptimizerKeeper { + private final ScheduledExecutorService keeperExecutor = + new ScheduledThreadPoolExecutor( + 1, + r -> { + Thread t = new Thread(r, "optimizer-keeper"); + t.setDaemon(true); + return t; + }); + + private Predicate<TaskRuntime<?>> taskRetryPredicate; + private java.util.function.Function<String, List<TaskRuntime<?>>> taskCollector; + private java.util.function.Consumer<TaskRuntime<?>> taskRetryHandler; + + void start( + Predicate<TaskRuntime<?>> taskRetryPredicate, + java.util.function.Function<String, List<TaskRuntime<?>>> taskCollector, + java.util.function.Consumer<TaskRuntime<?>> taskRetryHandler) { + this.taskRetryPredicate = taskRetryPredicate; + this.taskCollector = taskCollector; + this.taskRetryHandler = taskRetryHandler; + keeperExecutor.scheduleAtFixedRate(this::run, 0, 10, TimeUnit.SECONDS); + } + + void stop() { + keeperExecutor.shutdown(); + } + + private void run() { + try { + detectExpiredOptimizers(); + detectSuspendedTasks(); + } catch (Throwable t) { + LOG.error("OptimizerKeeper error", t); + } + } + + private void detectExpiredOptimizers() { + long now = System.currentTimeMillis(); + for (Map.Entry<String, OptimizerInstance> entry : authOptimizers.entrySet()) { + OptimizerInstance optimizer = entry.getValue(); + if (now - optimizer.getTouchTime() > optimizerTouchTimeout) { + LOG.warn( + "Optimizer {} from group {} has expired, removing", + optimizer.getResourceId(), + optimizer.getGroupName()); + unregisterOptimizer(entry.getKey()); + } + } + } + + private void detectSuspendedTasks() { + if (taskCollector == null || taskRetryPredicate == null || taskRetryHandler == null) { + return; + } + for (String group : quotaManager.getGroupNames()) { + try { + List<TaskRuntime<?>> tasks = taskCollector.apply(group); + if (tasks != null) { + tasks.stream() + .filter(taskRetryPredicate) + .forEach( + task -> { + LOG.warn("Resetting suspended task {} in group {}", task.getTaskId(), group); + taskRetryHandler.accept(task); + }); + } + } catch (Exception e) { + LOG.error("Error detecting suspended tasks for group {}", group, e); + } + } + } + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/resource/QuotaManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/resource/QuotaManager.java new file mode 100644 index 000000000..51fc2c999 --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/resource/QuotaManager.java @@ -0,0 +1,109 @@ +/* + * 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.resource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +/** Manages optimizer quotas per resource group. Thread-safe via internal lock. */ +public class QuotaManager implements QuotaProvider { + + private static final Logger LOG = LoggerFactory.getLogger(QuotaManager.class); + + private final Lock quotaLock = new ReentrantLock(); + private final Map<String, Set<OptimizerInstance>> optimizersByGroup = new HashMap<>(); + + @Override + public int getTotalQuota(String resourceGroup) { + quotaLock.lock(); + try { + Set<OptimizerInstance> optimizers = optimizersByGroup.get(resourceGroup); + if (optimizers == null || optimizers.isEmpty()) { + return 0; + } + return optimizers.stream().mapToInt(OptimizerInstance::getThreadCount).sum(); + } finally { + quotaLock.unlock(); + } + } + + public double getAvailableCore(String resourceGroup) { + return getTotalQuota(resourceGroup); + } + + public void addOptimizer(OptimizerInstance instance) { + quotaLock.lock(); + try { + optimizersByGroup + .computeIfAbsent(instance.getGroupName(), k -> new HashSet<>()) + .add(instance); + LOG.info( + "Added optimizer {} to group {}, total quota: {}", + instance.getResourceId(), + instance.getGroupName(), + getTotalQuota(instance.getGroupName())); + } finally { + quotaLock.unlock(); + } + } + + public void removeOptimizer(OptimizerInstance instance) { + quotaLock.lock(); + try { + Set<OptimizerInstance> optimizers = optimizersByGroup.get(instance.getGroupName()); + if (optimizers != null) { + optimizers.remove(instance); + if (optimizers.isEmpty()) { + optimizersByGroup.remove(instance.getGroupName()); + } + } + LOG.info( + "Removed optimizer {} from group {}", instance.getResourceId(), instance.getGroupName()); + } finally { + quotaLock.unlock(); + } + } + + public Set<OptimizerInstance> getOptimizers(String groupName) { + quotaLock.lock(); + try { + return Collections.unmodifiableSet( + optimizersByGroup.getOrDefault(groupName, Collections.emptySet())); + } finally { + quotaLock.unlock(); + } + } + + public Set<String> getGroupNames() { + quotaLock.lock(); + try { + return Collections.unmodifiableSet(optimizersByGroup.keySet()); + } finally { + quotaLock.unlock(); + } + } +} diff --git a/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ActionCoordinator old mode 100755 new mode 100644 similarity index 92% copy from amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory copy to amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ActionCoordinator index 96baa8ebf..a9061449d --- a/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory +++ b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ActionCoordinator @@ -16,4 +16,4 @@ # limitations under the License. # -org.apache.amoro.server.process.iceberg.IcebergProcessFactory +org.apache.amoro.server.optimizing.OptimizingActionCoordinator diff --git a/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory index 96baa8ebf..130687494 100755 --- a/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory +++ b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory @@ -17,3 +17,4 @@ # org.apache.amoro.server.process.iceberg.IcebergProcessFactory +org.apache.amoro.server.optimizing.IcebergOptimizingProcessFactory diff --git a/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingPlanResult.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingPlanResult.java new file mode 100644 index 000000000..d5a3cff79 --- /dev/null +++ b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingPlanResult.java @@ -0,0 +1,88 @@ +/* + * 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.optimizing; + +import org.apache.amoro.process.StagedTaskDescriptor; + +import java.util.List; +import java.util.Map; + +/** DTO carrying the result of an optimizing plan, including tasks and metadata. */ +public class OptimizingPlanResult { + + private final long processId; + private final OptimizingType optimizingType; + private final long planTime; + private final long targetSnapshotId; + private final long targetChangeSnapshotId; + private final List<? extends StagedTaskDescriptor<?, ?, ?>> tasks; + private final Map<String, Long> fromSequence; + private final Map<String, Long> toSequence; + + public OptimizingPlanResult( + long processId, + OptimizingType optimizingType, + long planTime, + long targetSnapshotId, + long targetChangeSnapshotId, + List<? extends StagedTaskDescriptor<?, ?, ?>> tasks, + Map<String, Long> fromSequence, + Map<String, Long> toSequence) { + this.processId = processId; + this.optimizingType = optimizingType; + this.planTime = planTime; + this.targetSnapshotId = targetSnapshotId; + this.targetChangeSnapshotId = targetChangeSnapshotId; + this.tasks = tasks; + this.fromSequence = fromSequence; + this.toSequence = toSequence; + } + + public long getProcessId() { + return processId; + } + + public OptimizingType getOptimizingType() { + return optimizingType; + } + + public long getPlanTime() { + return planTime; + } + + public long getTargetSnapshotId() { + return targetSnapshotId; + } + + public long getTargetChangeSnapshotId() { + return targetChangeSnapshotId; + } + + public List<? extends StagedTaskDescriptor<?, ?, ?>> getTasks() { + return tasks; + } + + public Map<String, Long> getFromSequence() { + return fromSequence; + } + + public Map<String, Long> getToSequence() { + return toSequence; + } +} diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingType.java similarity index 100% copy from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java copy to amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingType.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingCommitter.java similarity index 65% copy from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java copy to amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingCommitter.java index 2fbc75199..284d8daea 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java +++ b/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingCommitter.java @@ -18,8 +18,15 @@ package org.apache.amoro.optimizing; -public enum OptimizingType { - MINOR, - MAJOR, - FULL +import org.apache.amoro.exception.OptimizingCommitException; + +/** + * Internal abstraction for committing optimizing results, used by ProcessFactory implementations. + * Each format module provides its own committer implementation (e.g., UnKeyedTableCommit, + * KeyedTableCommit for Iceberg). + */ +public interface TableOptimizingCommitter { + + /** Commit the optimizing results to the table. */ + void commit() throws OptimizingCommitException; } diff --git a/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingPlanner.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingPlanner.java new file mode 100644 index 000000000..73f47196d --- /dev/null +++ b/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizingPlanner.java @@ -0,0 +1,56 @@ +/* + * 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.optimizing; + +import java.util.Map; + +/** + * Internal abstraction for table optimizing planning, used by ProcessFactory implementations. This + * is NOT a standalone SPI interface; ProcessFactory implementations encapsulate planner creation + * internally. + */ +public interface TableOptimizingPlanner { + + /** Evaluate whether optimizing is necessary for the current table state. */ + boolean isNecessary(); + + /** Execute the full planning and return the plan result including tasks. */ + OptimizingPlanResult plan(); + + /** Get the optimizing type determined by the planner. */ + OptimizingType getOptimizingType(); + + /** Get the unique process ID for this planning session. */ + long getProcessId(); + + /** Get the timestamp when planning started. */ + long getPlanTime(); + + /** Get the target snapshot ID that this plan is based on. */ + long getTargetSnapshotId(); + + /** Get the target change snapshot ID for keyed tables. */ + long getTargetChangeSnapshotId(); + + /** Get the from-sequence map for partition-level sequence tracking. */ + Map<String, Long> getFromSequence(); + + /** Get the to-sequence map for partition-level sequence tracking. */ + Map<String, Long> getToSequence(); +} diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java index 05bf82fab..0b7134da1 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractOptimizingPlanner.java @@ -21,8 +21,10 @@ package org.apache.amoro.optimizing.plan; import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.config.OptimizingConfig; import org.apache.amoro.iceberg.Constants; +import org.apache.amoro.optimizing.OptimizingPlanResult; import org.apache.amoro.optimizing.OptimizingType; import org.apache.amoro.optimizing.RewriteStageTask; +import org.apache.amoro.optimizing.TableOptimizingPlanner; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; import org.apache.amoro.table.KeyedTableSnapshot; import org.apache.amoro.table.MixedTable; @@ -43,7 +45,8 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; -public abstract class AbstractOptimizingPlanner extends AbstractOptimizingEvaluator { +public abstract class AbstractOptimizingPlanner extends AbstractOptimizingEvaluator + implements TableOptimizingPlanner { private static final Logger LOG = LoggerFactory.getLogger(AbstractOptimizingPlanner.class); private final Expression partitionFilter; @@ -219,4 +222,18 @@ public abstract class AbstractOptimizingPlanner extends AbstractOptimizingEvalua public long getProcessId() { return processId; } + + @Override + public OptimizingPlanResult plan() { + List<RewriteStageTask> tasks = planTasks(); + return new OptimizingPlanResult( + getProcessId(), + getOptimizingType(), + getPlanTime(), + getTargetSnapshotId(), + getTargetChangeSnapshotId(), + tasks, + getFromSequence(), + getToSequence()); + } } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonCompactionTask.java similarity index 57% copy from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java copy to amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonCompactionTask.java index 2fbc75199..b0ae73572 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java +++ b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonCompactionTask.java @@ -16,10 +16,30 @@ * limitations under the License. */ -package org.apache.amoro.optimizing; +package org.apache.amoro.formats.paimon.optimizing; -public enum OptimizingType { - MINOR, - MAJOR, - FULL +import org.apache.amoro.process.StagedTaskDescriptor; + +import java.util.Map; + +public class PaimonCompactionTask extends StagedTaskDescriptor<Void, Void, Void> { + + public PaimonCompactionTask() { + super(); + } + + public PaimonCompactionTask(long tableId, Void input, Map<String, String> properties) { + super(tableId, input, properties); + } + + @Override + protected void calculateSummary() { + // TODO: Implement + } + + @Override + protected Void deserializeOutput(byte[] outputBytes) { + // TODO: Implement + return null; + } } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingCommitter.java similarity index 75% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java rename to amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingCommitter.java index 2fbc75199..134e363b1 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingType.java +++ b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingCommitter.java @@ -16,10 +16,14 @@ * limitations under the License. */ -package org.apache.amoro.optimizing; +package org.apache.amoro.formats.paimon.optimizing; -public enum OptimizingType { - MINOR, - MAJOR, - FULL +import org.apache.amoro.optimizing.TableOptimizingCommitter; + +public class PaimonOptimizingCommitter implements TableOptimizingCommitter { + + @Override + public void commit() { + // TODO: Implement Paimon commit + } } diff --git a/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingPlanner.java b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingPlanner.java new file mode 100644 index 000000000..fb9e64fbf --- /dev/null +++ b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingPlanner.java @@ -0,0 +1,76 @@ +/* + * 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.formats.paimon.optimizing; + +import org.apache.amoro.optimizing.OptimizingPlanResult; +import org.apache.amoro.optimizing.OptimizingType; +import org.apache.amoro.optimizing.TableOptimizingPlanner; + +import java.util.Collections; +import java.util.Map; + +public class PaimonOptimizingPlanner implements TableOptimizingPlanner { + + @Override + public boolean isNecessary() { + // TODO: Implement Paimon optimization evaluation + return false; + } + + @Override + public OptimizingPlanResult plan() { + // TODO: Implement Paimon planning + return null; + } + + @Override + public OptimizingType getOptimizingType() { + return null; + } + + @Override + public long getProcessId() { + return 0; + } + + @Override + public long getPlanTime() { + return 0; + } + + @Override + public long getTargetSnapshotId() { + return 0; + } + + @Override + public long getTargetChangeSnapshotId() { + return 0; + } + + @Override + public Map<String, Long> getFromSequence() { + return Collections.emptyMap(); + } + + @Override + public Map<String, Long> getToSequence() { + return Collections.emptyMap(); + } +} diff --git a/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingProcessFactory.java b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingProcessFactory.java new file mode 100644 index 000000000..5d2d281ad --- /dev/null +++ b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/optimizing/PaimonOptimizingProcessFactory.java @@ -0,0 +1,74 @@ +/* + * 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.formats.paimon.optimizing; + +import org.apache.amoro.Action; +import org.apache.amoro.TableFormat; +import org.apache.amoro.TableRuntime; +import org.apache.amoro.process.ProcessFactory; +import org.apache.amoro.process.ProcessTriggerStrategy; +import org.apache.amoro.process.RecoverProcessFailedException; +import org.apache.amoro.process.TableProcess; +import org.apache.amoro.process.TableProcessStore; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +public class PaimonOptimizingProcessFactory implements ProcessFactory { + + @Override + public String name() { + return "paimon-optimizing-process-factory"; + } + + @Override + public Map<TableFormat, Set<Action>> supportedActions() { + return new HashMap<>(); + } + + @Override + public ProcessTriggerStrategy triggerStrategy(TableFormat format, Action action) { + return ProcessTriggerStrategy.METADATA_TRIGGER; + } + + @Override + public Optional<TableProcess> trigger(TableRuntime tableRuntime, Action action) { + // TODO: Implement Paimon compaction + return Optional.empty(); + } + + @Override + public TableProcess recover(TableRuntime tableRuntime, TableProcessStore store) + throws RecoverProcessFailedException { + // TODO: Implement recovery + throw new RecoverProcessFailedException("Paimon process recovery not yet implemented"); + } + + @Override + public void open(Map<String, String> properties) { + // No initialization needed + } + + @Override + public void close() { + // No cleanup needed + } +}
