Skip to content

[AMORO-4208] Refactor orphan-files-cleaning via ProcessFactory plugin #4209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhoujinsong merged 3 commits into from
May 14, 2026
Merged

[AMORO-4208] Refactor orphan-files-cleaning via ProcessFactory plugin #4209

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ded0027
Refactor orphan-files-cleaning via ProcessFactory plugin
May 7, 2026
e7eed31
Remove unused orphan file cleaning configs and inline executor
May 14, 2026
d469288
Add process factory and execute engine configuration docs
May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 0 additions & 18 deletions amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,24 +174,6 @@ public class AmoroManagementConf {
.defaultValue(Duration.ofHours(1))
.withDescription("Interval for expiring snapshots.");

public static final ConfigOption<Boolean> CLEAN_ORPHAN_FILES_ENABLED =
ConfigOptions.key("clean-orphan-files.enabled")
.booleanType()
.defaultValue(true)
.withDescription("Enable orphan files cleaning.");

public static final ConfigOption<Integer> CLEAN_ORPHAN_FILES_THREAD_COUNT =
ConfigOptions.key("clean-orphan-files.thread-count")
.intType()
.defaultValue(10)
.withDescription("The number of threads used for orphan files cleaning.");

public static final ConfigOption<Duration> CLEAN_ORPHAN_FILES_INTERVAL =
ConfigOptions.key("clean-orphan-files.interval")
.durationType()
.defaultValue(Duration.ofDays(1))
.withDescription("Interval for cleaning orphan files.");

public static final ConfigOption<Boolean> CLEAN_DANGLING_DELETE_FILES_ENABLED =
ConfigOptions.key("clean-dangling-delete-files.enabled")
.booleanType()
Expand Down
3 changes: 0 additions & 3 deletions amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConfValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,9 +86,6 @@ public static void validateConfig(Configurations configurations) {
validateThreadCount(configurations, AmoroManagementConf.EXPIRE_SNAPSHOTS_THREAD_COUNT);
}

if (configurations.getBoolean(AmoroManagementConf.CLEAN_ORPHAN_FILES_ENABLED)) {
validateThreadCount(configurations, AmoroManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT);
}
if (configurations.getBoolean(AmoroManagementConf.SYNC_HIVE_TABLES_ENABLED)) {
validateThreadCount(configurations, AmoroManagementConf.SYNC_HIVE_TABLES_THREAD_COUNT);
}
Expand Down
1 change: 0 additions & 1 deletion amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,6 @@ public void startOptimizingService() throws Exception {
addHandlerChain(optimizingService.getTableRuntimeHandler());
addHandlerChain(processService.getTableHandlerChain());
addHandlerChain(InlineTableExecutors.getInstance().getDataExpiringExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getOptimizingCommitExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getProcessDataExpiringExecutor());
Expand Down
32 changes: 32 additions & 0 deletions amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,14 @@ public class IcebergProcessFactory implements ProcessFactory {
.durationType()
.defaultValue(Duration.ofHours(1));

public static final ConfigOption<Boolean> ORPHAN_FILES_CLEANING_ENABLED =
ConfigOptions.key("clean-orphan-files.enabled").booleanType().defaultValue(true);

public static final ConfigOption<Duration> ORPHAN_FILES_CLEANING_INTERVAL =
ConfigOptions.key("clean-orphan-files.interval")
.durationType()
.defaultValue(Duration.ofDays(1));

private ExecuteEngine localEngine;
private final Map<Action, ProcessTriggerStrategy> actions = Maps.newHashMap();
private final List<TableFormat> formats =
Expand Down Expand Up @@ -91,7 +99,10 @@ public Optional<TableProcess> trigger(TableRuntime tableRuntime, Action action)

if (IcebergActions.EXPIRE_SNAPSHOTS.equals(action)) {
return triggerExpireSnapshot(tableRuntime);
} else if (IcebergActions.DELETE_ORPHANS.equals(action)) {
return triggerCleanOrphans(tableRuntime);
}

return Optional.empty();
}

Expand All @@ -113,6 +124,12 @@ public void open(Map<String, String> properties) {
this.actions.put(
IcebergActions.EXPIRE_SNAPSHOTS, ProcessTriggerStrategy.triggerAtFixRate(interval));
}

if (configs.getBoolean(ORPHAN_FILES_CLEANING_ENABLED)) {
Duration interval = configs.getDuration(ORPHAN_FILES_CLEANING_INTERVAL);
this.actions.put(
IcebergActions.DELETE_ORPHANS, ProcessTriggerStrategy.triggerAtFixRate(interval));
}
}

private Optional<TableProcess> triggerExpireSnapshot(TableRuntime tableRuntime) {
Expand All @@ -130,6 +147,21 @@ private Optional<TableProcess> triggerExpireSnapshot(TableRuntime tableRuntime)
return Optional.of(new SnapshotsExpiringProcess(tableRuntime, localEngine));
}

private Optional<TableProcess> triggerCleanOrphans(TableRuntime tableRuntime) {
if (localEngine == null || !tableRuntime.getTableConfiguration().isCleanOrphanEnabled()) {
return Optional.empty();
}

long lastExecuteTime =
tableRuntime.getState(DefaultTableRuntime.CLEANUP_STATE_KEY).getLastOrphanFilesCleanTime();
ProcessTriggerStrategy strategy = actions.get(IcebergActions.DELETE_ORPHANS);
if (System.currentTimeMillis() - lastExecuteTime < strategy.getTriggerInterval().toMillis()) {
return Optional.empty();
}

return Optional.of(new OrphanFilesCleaningProcess(tableRuntime, localEngine));
}

@Override
public void close() {}

Expand Down
79 changes: 79 additions & 0 deletions ...ams/src/main/java/org/apache/amoro/server/process/iceberg/OrphanFilesCleaningProcess.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
/*
* 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.process.iceberg;

import org.apache.amoro.Action;
import org.apache.amoro.AmoroTable;
import org.apache.amoro.IcebergActions;
import org.apache.amoro.TableRuntime;
import org.apache.amoro.maintainer.TableMaintainer;
import org.apache.amoro.process.ExecuteEngine;
import org.apache.amoro.process.LocalProcess;
import org.apache.amoro.process.TableProcess;
import org.apache.amoro.server.optimizing.maintainer.TableMaintainerFactory;
import org.apache.amoro.server.table.DefaultTableRuntime;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/** Local table process for cleaning Iceberg orphan files. */
public class OrphanFilesCleaningProcess extends TableProcess implements LocalProcess {

private static final Logger LOG = LoggerFactory.getLogger(OrphanFilesCleaningProcess.class);

public OrphanFilesCleaningProcess(TableRuntime tableRuntime, ExecuteEngine engine) {
super(tableRuntime, engine);
}

@Override
public String tag() {
return getAction().getName().toLowerCase();
}

@Override
public void run() {
try {
AmoroTable<?> amoroTable = tableRuntime.loadTable();
TableMaintainer tableMaintainer = TableMaintainerFactory.create(amoroTable, tableRuntime);
tableMaintainer.cleanOrphanFiles();
tableRuntime.updateState(
DefaultTableRuntime.CLEANUP_STATE_KEY,
cleanUp -> cleanUp.setLastOrphanFilesCleanTime(System.currentTimeMillis()));
} catch (Throwable t) {
LOG.error("Failed to clean orphan files for table {}", tableRuntime.getTableIdentifier(), t);
}
}

@Override
public Action getAction() {
return IcebergActions.DELETE_ORPHANS;
}

@Override
public Map<String, String> getProcessParameters() {
return Maps.newHashMap();
}

@Override
public Map<String, String> getSummary() {
return Maps.newHashMap();
}
}
12 changes: 0 additions & 12 deletions amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ public class InlineTableExecutors {

private static final InlineTableExecutors instance = new InlineTableExecutors();
private TableRuntimeRefreshExecutor tableRefreshingExecutor;
private OrphanFilesCleaningExecutor orphanFilesCleaningExecutor;
private DanglingDeleteFilesCleaningExecutor danglingDeleteFilesCleaningExecutor;
private BlockerExpiringExecutor blockerExpiringExecutor;
private OptimizingCommitExecutor optimizingCommitExecutor;
Expand All @@ -42,13 +41,6 @@ public static InlineTableExecutors getInstance() {
}
Copy link
Copy Markdown
Contributor

@zhoujinsong zhoujinsong May 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The old global configs in AmoroManagementConf are still present after this refactor:

// AmoroManagementConf.java
public static final ConfigOption<Boolean> CLEAN_ORPHAN_FILES_ENABLED = ...
public static final ConfigOption<Integer> CLEAN_ORPHAN_FILES_THREAD_COUNT = ...
public static final ConfigOption<Duration> CLEAN_ORPHAN_FILES_INTERVAL = ...

And AmoroManagementConfValidator still validates them. Since the configuration has moved to process-factories.yaml, these entries are now dead code. They should either be removed or marked @Deprecated with a note pointing to the new config location, to avoid confusing users who upgrade and wonder why their old ams.yaml settings are silently ignored.

Copy link
Copy Markdown
Contributor Author

@zhangwl9 zhangwl9 May 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the review! The old CLEAN_ORPHAN_FILES_* configs in AmoroManagementConf and their validation in AmoroManagementConfValidator have already been removed in the follow-up commit . The configuration has been fully migrated to IcebergProcessFactory (process-factories.yaml) and documented in the deployment guide.


public void setup(TableService tableService, Configurations conf) {
if (conf.getBoolean(AmoroManagementConf.CLEAN_ORPHAN_FILES_ENABLED)) {
this.orphanFilesCleaningExecutor =
new OrphanFilesCleaningExecutor(
tableService,
conf.getInteger(AmoroManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT),
conf.get(AmoroManagementConf.CLEAN_ORPHAN_FILES_INTERVAL));
}
if (conf.getBoolean(AmoroManagementConf.CLEAN_DANGLING_DELETE_FILES_ENABLED)) {
this.danglingDeleteFilesCleaningExecutor =
new DanglingDeleteFilesCleaningExecutor(
Expand Down Expand Up @@ -108,10 +100,6 @@ public TableRuntimeRefreshExecutor getTableRefreshingExecutor() {
return tableRefreshingExecutor;
}

public OrphanFilesCleaningExecutor getOrphanFilesCleaningExecutor() {
return orphanFilesCleaningExecutor;
}

public DanglingDeleteFilesCleaningExecutor getDanglingDeleteFilesCleaningExecutor() {
return danglingDeleteFilesCleaningExecutor;
}
Expand Down
85 changes: 0 additions & 85 deletions ...s/src/main/java/org/apache/amoro/server/scheduler/inline/OrphanFilesCleaningExecutor.java
View file
Open in desktop

This file was deleted.

5 changes: 0 additions & 5 deletions amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,8 +342,6 @@ public void beginProcess(OptimizingProcess optimizingProcess) {
public long getLastCleanTime(CleanupOperation operation) {
TableRuntimeCleanupState state = store().getState(CLEANUP_STATE_KEY);
switch (operation) {
case ORPHAN_FILES_CLEANING:
return state.getLastOrphanFilesCleanTime();
case DANGLING_DELETE_FILES_CLEANING:
return state.getLastDanglingDeleteFilesCleanTime();
case DATA_EXPIRING:
Expand All @@ -362,9 +360,6 @@ public void updateLastCleanTime(CleanupOperation operation, long time) {
CLEANUP_STATE_KEY,
state -> {
switch (operation) {
case ORPHAN_FILES_CLEANING:
state.setLastOrphanFilesCleanTime(time);
break;
case DANGLING_DELETE_FILES_CLEANING:
state.setLastDanglingDeleteFilesCleanTime(time);
break;
Expand Down
1 change: 0 additions & 1 deletion amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
/** Table cleanup operation enum. Defines different operation types for table cleanup tasks. */
public enum CleanupOperation {
DANGLING_DELETE_FILES_CLEANING,
ORPHAN_FILES_CLEANING,
DATA_EXPIRING,
SNAPSHOTS_EXPIRING,
// NONE indicates operation types where no cleanup process records are
Expand Down
3 changes: 2 additions & 1 deletion amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,9 @@ public long getLastOrphanFilesCleanTime() {
return lastOrphanFilesCleanTime;
}

public void setLastOrphanFilesCleanTime(long lastOrphanFilesCleanTime) {
public TableRuntimeCleanupState setLastOrphanFilesCleanTime(long lastOrphanFilesCleanTime) {
this.lastOrphanFilesCleanTime = lastOrphanFilesCleanTime;
return this;
}

public long getLastDanglingDeleteFilesCleanTime() {
Expand Down
1 change: 0 additions & 1 deletion amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,6 @@ private String getAmsConfig() {
+ " refresh-table-thread-count: 10\n"
+ " refresh-table-interval: 60000 #1min\n"
+ " expire-table-thread-count: 10\n"
+ " clean-orphan-file-thread-count: 10\n"
+ " sync-hive-tables-thread-count: 10\n"
+ "\n"
+ " thrift-server:\n"
Expand Down
8 changes: 0 additions & 8 deletions amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConfValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,14 +135,6 @@ public void testValidateThreadCount() {
configurations.setInteger(AmoroManagementConf.EXPIRE_SNAPSHOTS_THREAD_COUNT, 10);
AmoroManagementConfValidator.validateConfig(configurations);

configurations.setBoolean(AmoroManagementConf.CLEAN_ORPHAN_FILES_ENABLED, true);
configurations.setInteger(AmoroManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT, -1);
Assert.assertThrows(
IllegalArgumentException.class,
() -> AmoroManagementConfValidator.validateConfig(configurations));
configurations.setInteger(AmoroManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT, 10);
AmoroManagementConfValidator.validateConfig(configurations);

configurations.setBoolean(AmoroManagementConf.SYNC_HIVE_TABLES_ENABLED, true);
configurations.setInteger(AmoroManagementConf.SYNC_HIVE_TABLES_THREAD_COUNT, -1);
Assert.assertThrows(
Expand Down
Loading
Loading