This is an automated email from the ASF dual-hosted git repository. czy006 pushed a commit to branch 0.9.x in repository https://gitbox.apache.org/repos/asf/amoro.git
commit fc734b5eb7d86035614e41e437c62e8171fe1c9f Author: WenLingzhang <[email protected]> AuthorDate: Thu May 14 15:07:50 2026 +0800 [AMORO-4208] Refactor orphan-files-cleaning via ProcessFactory plugin (#4209) * Refactor orphan-files-cleaning via ProcessFactory plugin # Conflicts: # amoro-common/src/test/java/org/apache/amoro/process/TestLocalExecutionEngine.java * Remove unused orphan file cleaning configs and inline executor * Add process factory and execute engine configuration docs --------- Co-authored-by: 张文领 <[email protected]> (cherry picked from commit 75dd6570c0d8125bf892a2747342a16df089351d) --- .../apache/amoro/server/AmoroManagementConf.java | 18 --- .../amoro/server/AmoroManagementConfValidator.java | 3 - .../apache/amoro/server/AmoroServiceContainer.java | 1 - .../process/iceberg/IcebergProcessFactory.java | 32 ++++++ .../iceberg/OrphanFilesCleaningProcess.java | 79 +++++++++++++ .../scheduler/inline/InlineTableExecutors.java | 12 -- .../inline/OrphanFilesCleaningExecutor.java | 85 -------------- .../amoro/server/table/DefaultTableRuntime.java | 5 - .../server/table/cleanup/CleanupOperation.java | 1 - .../table/cleanup/TableRuntimeCleanupState.java | 3 +- .../org/apache/amoro/server/AmsEnvironment.java | 1 - .../server/TestAmoroManagementConfValidator.java | 8 -- .../process/iceberg/TestIcebergProcessFactory.java | 128 ++++++++++++--------- .../inline/PeriodicTableSchedulerTestBase.java | 3 - .../inline/TestPeriodicTableSchedulerCleanup.java | 3 - .../amoro/process/TestLocalExecutionEngine.java | 10 +- .../iceberg/maintainer/IcebergTableMaintainer.java | 2 +- charts/amoro/templates/amoro-configmap.yaml | 4 - dist/src/main/amoro-bin/conf/config.yaml | 5 - .../amoro-bin/conf/plugins/execute-engines.yaml | 1 + .../amoro-bin/conf/plugins/process-factories.yaml | 2 + docs/admin-guides/deployment.md | 64 +++++++++++ docs/configuration/ams-config.md | 3 - 23 files changed, 265 insertions(+), 208 deletions(-) diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java index 9ac6fb08e..49eade550 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java @@ -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() diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConfValidator.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConfValidator.java index 1e78520f0..627b577d6 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConfValidator.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConfValidator.java @@ -86,9 +86,6 @@ public class AmoroManagementConfValidator { 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); } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index 0be364523..d174e559e 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -290,7 +290,6 @@ public class AmoroServiceContainer { 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()); diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java index 18de241db..1f7813476 100755 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java @@ -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 = @@ -91,7 +99,10 @@ public class IcebergProcessFactory implements ProcessFactory { if (IcebergActions.EXPIRE_SNAPSHOTS.equals(action)) { return triggerExpireSnapshot(tableRuntime); + } else if (IcebergActions.DELETE_ORPHANS.equals(action)) { + return triggerCleanOrphans(tableRuntime); } + return Optional.empty(); } @@ -113,6 +124,12 @@ public class IcebergProcessFactory implements ProcessFactory { 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) { @@ -130,6 +147,21 @@ public class IcebergProcessFactory implements ProcessFactory { 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() {} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/OrphanFilesCleaningProcess.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/OrphanFilesCleaningProcess.java new file mode 100644 index 000000000..22e4c05ee --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/OrphanFilesCleaningProcess.java @@ -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(); + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java index 173c0259e..4e5b4beb3 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java @@ -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; @@ -42,13 +41,6 @@ public class InlineTableExecutors { } 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( @@ -108,10 +100,6 @@ public class InlineTableExecutors { return tableRefreshingExecutor; } - public OrphanFilesCleaningExecutor getOrphanFilesCleaningExecutor() { - return orphanFilesCleaningExecutor; - } - public DanglingDeleteFilesCleaningExecutor getDanglingDeleteFilesCleaningExecutor() { return danglingDeleteFilesCleaningExecutor; } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OrphanFilesCleaningExecutor.java b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OrphanFilesCleaningExecutor.java deleted file mode 100644 index 19db0f192..000000000 --- a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OrphanFilesCleaningExecutor.java +++ /dev/null @@ -1,85 +0,0 @@ -/* - * 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.scheduler.inline; - -import org.apache.amoro.AmoroTable; -import org.apache.amoro.TableRuntime; -import org.apache.amoro.config.TableConfiguration; -import org.apache.amoro.maintainer.TableMaintainer; -import org.apache.amoro.server.optimizing.maintainer.TableMaintainers; -import org.apache.amoro.server.scheduler.PeriodicTableScheduler; -import org.apache.amoro.server.table.TableService; -import org.apache.amoro.server.table.cleanup.CleanupOperation; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.time.Duration; -import java.util.concurrent.ThreadLocalRandom; - -public class OrphanFilesCleaningExecutor extends PeriodicTableScheduler { - private static final Logger LOG = LoggerFactory.getLogger(OrphanFilesCleaningExecutor.class); - private final Duration interval; - - public OrphanFilesCleaningExecutor(TableService tableService, int poolSize, Duration interval) { - super(tableService, poolSize); - this.interval = interval; - } - - @Override - protected long getNextExecutingTime(TableRuntime tableRuntime) { - return interval.toMillis(); - } - - @Override - protected boolean shouldExecute(Long lastCleanupEndTime) { - return System.currentTimeMillis() - lastCleanupEndTime >= interval.toMillis(); - } - - @Override - protected CleanupOperation getCleanupOperation() { - return CleanupOperation.ORPHAN_FILES_CLEANING; - } - - @Override - protected boolean enabled(TableRuntime tableRuntime) { - return tableRuntime.getTableConfiguration().isCleanOrphanEnabled(); - } - - @Override - public void handleConfigChanged(TableRuntime tableRuntime, TableConfiguration originalConfig) { - scheduleIfNecessary(tableRuntime, getStartDelay()); - } - - @Override - protected long getExecutorDelay() { - return ThreadLocalRandom.current().nextLong(interval.toMillis()); - } - - @Override - public void execute(TableRuntime tableRuntime) { - try { - LOG.info("{} start cleaning orphan files", tableRuntime.getTableIdentifier()); - AmoroTable<?> amoroTable = loadTable(tableRuntime); - TableMaintainer tableMaintainer = TableMaintainers.create(amoroTable, tableRuntime); - tableMaintainer.cleanOrphanFiles(); - } catch (Throwable t) { - LOG.error("{} failed to clean orphan file", tableRuntime.getTableIdentifier(), t); - } - } -} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java index 9b4b56a95..f4a4ef73e 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java @@ -342,8 +342,6 @@ public class DefaultTableRuntime extends AbstractTableRuntime { 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: @@ -362,9 +360,6 @@ public class DefaultTableRuntime extends AbstractTableRuntime { CLEANUP_STATE_KEY, state -> { switch (operation) { - case ORPHAN_FILES_CLEANING: - state.setLastOrphanFilesCleanTime(time); - break; case DANGLING_DELETE_FILES_CLEANING: state.setLastDanglingDeleteFilesCleanTime(time); break; diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java index 10afefe63..b6597db82 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java @@ -21,7 +21,6 @@ package org.apache.amoro.server.table.cleanup; /** 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 diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java index f65ae387c..639506ea9 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java @@ -28,8 +28,9 @@ public class TableRuntimeCleanupState { return lastOrphanFilesCleanTime; } - public void setLastOrphanFilesCleanTime(long lastOrphanFilesCleanTime) { + public TableRuntimeCleanupState setLastOrphanFilesCleanTime(long lastOrphanFilesCleanTime) { this.lastOrphanFilesCleanTime = lastOrphanFilesCleanTime; + return this; } public long getLastDanglingDeleteFilesCleanTime() { diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java index 93fd182d3..916765d26 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java @@ -369,7 +369,6 @@ public class AmsEnvironment { + " 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" diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConfValidator.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConfValidator.java index 8e52f779b..532bc1888 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConfValidator.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConfValidator.java @@ -135,14 +135,6 @@ public class TestAmoroManagementConfValidator { 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( diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java index 35151ce3d..61b00b833 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java @@ -43,104 +43,128 @@ public class TestIcebergProcessFactory { @Test public void testOpenAndSupportedActions() { + assertSupportedAction("expire-snapshots", IcebergActions.EXPIRE_SNAPSHOTS, Duration.ofHours(1)); + assertSupportedAction( + "clean-orphan-files", IcebergActions.DELETE_ORPHANS, Duration.ofHours(24)); + } + + @Test + public void testTriggerActionWhenDue() { + assertTriggerWhenDue( + "expire-snapshots", IcebergActions.EXPIRE_SNAPSHOTS, SnapshotsExpiringProcess.class, 0); + assertTriggerWhenDue( + "clean-orphan-files", IcebergActions.DELETE_ORPHANS, OrphanFilesCleaningProcess.class, 0); + } + + @Test + public void testTriggerActionNotDue() { + assertTriggerNotDue( + "expire-snapshots", IcebergActions.EXPIRE_SNAPSHOTS, System.currentTimeMillis()); + assertTriggerNotDue( + "clean-orphan-files", IcebergActions.DELETE_ORPHANS, System.currentTimeMillis()); + } + + @Test + public void testTriggerActionDisabled() { + assertTriggerDisabled("expire-snapshots", IcebergActions.EXPIRE_SNAPSHOTS, false, 0); + assertTriggerDisabled("clean-orphan-files", IcebergActions.DELETE_ORPHANS, false, 0); + } + + private void assertSupportedAction( + String configKey, org.apache.amoro.Action action, Duration interval) { IcebergProcessFactory factory = new IcebergProcessFactory(); Map<String, String> properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.interval", "1h"); + properties.put(configKey + ".enabled", "true"); + properties.put(configKey + ".interval", interval.toHours() + "h"); factory.open(properties); Map<TableFormat, Set<org.apache.amoro.Action>> supported = factory.supportedActions(); - Assert.assertTrue(supported.get(TableFormat.ICEBERG).contains(IcebergActions.EXPIRE_SNAPSHOTS)); - Assert.assertTrue( - supported.get(TableFormat.MIXED_ICEBERG).contains(IcebergActions.EXPIRE_SNAPSHOTS)); - Assert.assertTrue( - supported.get(TableFormat.MIXED_HIVE).contains(IcebergActions.EXPIRE_SNAPSHOTS)); - - ProcessTriggerStrategy strategy = - factory.triggerStrategy(TableFormat.ICEBERG, IcebergActions.EXPIRE_SNAPSHOTS); - Assert.assertEquals(Duration.ofHours(1), strategy.getTriggerInterval()); + Assert.assertTrue(supported.get(TableFormat.ICEBERG).contains(action)); + Assert.assertTrue(supported.get(TableFormat.MIXED_ICEBERG).contains(action)); + Assert.assertTrue(supported.get(TableFormat.MIXED_HIVE).contains(action)); + + ProcessTriggerStrategy strategy = factory.triggerStrategy(TableFormat.ICEBERG, action); + Assert.assertEquals(interval, strategy.getTriggerInterval()); } - @Test - public void testTriggerExpireSnapshotWhenDue() { + private void assertTriggerWhenDue( + String configKey, org.apache.amoro.Action action, Class<?> processClass, long lastTime) { IcebergProcessFactory factory = new IcebergProcessFactory(); Map<String, String> properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.interval", "1h"); + properties.put(configKey + ".enabled", "true"); + properties.put(configKey + ".interval", "1h"); factory.open(properties); LocalExecutionEngine localEngine = mock(LocalExecutionEngine.class); doReturn(LocalExecutionEngine.ENGINE_NAME).when(localEngine).name(); factory.availableExecuteEngines(Arrays.asList(localEngine)); - TableConfiguration tableConfiguration = new TableConfiguration().setExpireSnapshotEnabled(true); - TableRuntimeCleanupState cleanupState = - new TableRuntimeCleanupState().setLastSnapshotsExpiringTime(0); + TableRuntime runtime = createRuntime(configKey, true, lastTime); - TableRuntime runtime = mock(TableRuntime.class); - doReturn(tableConfiguration).when(runtime).getTableConfiguration(); - doReturn(cleanupState).when(runtime).getState(DefaultTableRuntime.CLEANUP_STATE_KEY); - - Optional<org.apache.amoro.process.TableProcess> process = - factory.trigger(runtime, IcebergActions.EXPIRE_SNAPSHOTS); + Optional<org.apache.amoro.process.TableProcess> process = factory.trigger(runtime, action); Assert.assertTrue(process.isPresent()); - Assert.assertTrue(process.get() instanceof SnapshotsExpiringProcess); + Assert.assertTrue(processClass.isInstance(process.get())); Assert.assertEquals(LocalExecutionEngine.ENGINE_NAME, process.get().getExecutionEngine()); } - @Test - public void testTriggerExpireSnapshotNotDue() { + private void assertTriggerNotDue( + String configKey, org.apache.amoro.Action action, long lastTime) { IcebergProcessFactory factory = new IcebergProcessFactory(); Map<String, String> properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.interval", "1h"); + properties.put(configKey + ".enabled", "true"); + properties.put(configKey + ".interval", "1h"); factory.open(properties); factory.availableExecuteEngines(Arrays.asList(mock(LocalExecutionEngine.class))); - TableConfiguration tableConfiguration = new TableConfiguration().setExpireSnapshotEnabled(true); - long now = System.currentTimeMillis(); - TableRuntimeCleanupState cleanupState = - new TableRuntimeCleanupState().setLastSnapshotsExpiringTime(now); - - TableRuntime runtime = mock(TableRuntime.class); - doReturn(tableConfiguration).when(runtime).getTableConfiguration(); - doReturn(cleanupState).when(runtime).getState(DefaultTableRuntime.CLEANUP_STATE_KEY); + TableRuntime runtime = createRuntime(configKey, true, lastTime); - Optional<org.apache.amoro.process.TableProcess> process = - factory.trigger(runtime, IcebergActions.EXPIRE_SNAPSHOTS); + Optional<org.apache.amoro.process.TableProcess> process = factory.trigger(runtime, action); Assert.assertFalse(process.isPresent()); } - @Test - public void testTriggerExpireSnapshotDisabled() { + private void assertTriggerDisabled( + String configKey, org.apache.amoro.Action action, boolean enabled, long lastTime) { IcebergProcessFactory factory = new IcebergProcessFactory(); Map<String, String> properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.interval", "1h"); + properties.put(configKey + ".enabled", "true"); + properties.put(configKey + ".interval", "1h"); factory.open(properties); factory.availableExecuteEngines(Arrays.asList(mock(LocalExecutionEngine.class))); - TableConfiguration tableConfiguration = - new TableConfiguration().setExpireSnapshotEnabled(false); - TableRuntimeCleanupState cleanupState = - new TableRuntimeCleanupState().setLastSnapshotsExpiringTime(0); + TableRuntime runtime = createRuntime(configKey, enabled, lastTime); + + Optional<org.apache.amoro.process.TableProcess> process = factory.trigger(runtime, action); + + Assert.assertFalse(process.isPresent()); + } + + private TableRuntime createRuntime(String configKey, boolean enabled, long lastTime) { + TableConfiguration tableConfiguration = new TableConfiguration(); + if ("expire-snapshots".equals(configKey)) { + tableConfiguration.setExpireSnapshotEnabled(enabled); + } else if ("clean-orphan-files".equals(configKey)) { + tableConfiguration.setCleanOrphanEnabled(enabled); + } + + TableRuntimeCleanupState cleanupState = new TableRuntimeCleanupState(); + if ("expire-snapshots".equals(configKey)) { + cleanupState.setLastSnapshotsExpiringTime(lastTime); + } else if ("clean-orphan-files".equals(configKey)) { + cleanupState.setLastOrphanFilesCleanTime(lastTime); + } TableRuntime runtime = mock(TableRuntime.class); doReturn(tableConfiguration).when(runtime).getTableConfiguration(); doReturn(cleanupState).when(runtime).getState(DefaultTableRuntime.CLEANUP_STATE_KEY); - - Optional<org.apache.amoro.process.TableProcess> process = - factory.trigger(runtime, IcebergActions.EXPIRE_SNAPSHOTS); - - Assert.assertFalse(process.isPresent()); + return runtime; } } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java index 50c0e9d19..ba2a8767c 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java @@ -31,7 +31,6 @@ class PeriodicTableSchedulerTestBase extends PeriodicTableScheduler { private final CleanupOperation cleanupOperation; private final boolean enabled; private static final long SNAPSHOTS_EXPIRING_INTERVAL = 60 * 60 * 1000L; // 1 hour - private static final long ORPHAN_FILES_CLEANING_INTERVAL = 24 * 60 * 60 * 1000L; // 1 day private static final long DANGLING_DELETE_FILES_CLEANING_INTERVAL = 24 * 60 * 60 * 1000L; private static final long DATA_EXPIRING_INTERVAL = 60 * 60 * 1000L; // 1 hour @@ -73,8 +72,6 @@ class PeriodicTableSchedulerTestBase extends PeriodicTableScheduler { switch (cleanupOperation) { case SNAPSHOTS_EXPIRING: return currentTime - lastCleanupEndTime >= SNAPSHOTS_EXPIRING_INTERVAL; - case ORPHAN_FILES_CLEANING: - return currentTime - lastCleanupEndTime >= ORPHAN_FILES_CLEANING_INTERVAL; case DANGLING_DELETE_FILES_CLEANING: return currentTime - lastCleanupEndTime >= DANGLING_DELETE_FILES_CLEANING_INTERVAL; case DATA_EXPIRING: diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestPeriodicTableSchedulerCleanup.java b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestPeriodicTableSchedulerCleanup.java index de25eac21..fef2a3bd4 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestPeriodicTableSchedulerCleanup.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestPeriodicTableSchedulerCleanup.java @@ -167,7 +167,6 @@ public class TestPeriodicTableSchedulerCleanup extends PersistentBase { public void testShouldExecuteTaskWithNoPreviousCleanup() { List<CleanupOperation> operations = Arrays.asList( - CleanupOperation.ORPHAN_FILES_CLEANING, CleanupOperation.DANGLING_DELETE_FILES_CLEANING, CleanupOperation.DATA_EXPIRING, CleanupOperation.SNAPSHOTS_EXPIRING); @@ -192,7 +191,6 @@ public class TestPeriodicTableSchedulerCleanup extends PersistentBase { public void testShouldNotExecuteTaskWithRecentCleanup() { List<CleanupOperation> operations = Arrays.asList( - CleanupOperation.ORPHAN_FILES_CLEANING, CleanupOperation.DANGLING_DELETE_FILES_CLEANING, CleanupOperation.DATA_EXPIRING, CleanupOperation.SNAPSHOTS_EXPIRING); @@ -222,7 +220,6 @@ public class TestPeriodicTableSchedulerCleanup extends PersistentBase { public void testShouldExecuteTaskWithOldCleanup() { List<CleanupOperation> operations = Arrays.asList( - CleanupOperation.ORPHAN_FILES_CLEANING, CleanupOperation.DANGLING_DELETE_FILES_CLEANING, CleanupOperation.DATA_EXPIRING, CleanupOperation.SNAPSHOTS_EXPIRING); diff --git a/amoro-common/src/test/java/org/apache/amoro/process/TestLocalExecutionEngine.java b/amoro-common/src/test/java/org/apache/amoro/process/TestLocalExecutionEngine.java index 6c67a2097..ed1b26d53 100644 --- a/amoro-common/src/test/java/org/apache/amoro/process/TestLocalExecutionEngine.java +++ b/amoro-common/src/test/java/org/apache/amoro/process/TestLocalExecutionEngine.java @@ -45,6 +45,11 @@ public class TestLocalExecutionEngine { @Test public void testSubmitUsesCustomPoolByTag() throws Exception { + assertCustomPoolByTag("snapshots-expiring"); + assertCustomPoolByTag("orphan-files-cleaning"); + } + + private void assertCustomPoolByTag(String tag) throws Exception { engine = createEngineWithTtl("1h"); CountDownLatch started = new CountDownLatch(1); @@ -54,7 +59,7 @@ public class TestLocalExecutionEngine { new LocalProcessTableProcess( mock(TableRuntime.class), engine, - "snapshots-expiring", + tag, () -> { threadName.set(Thread.currentThread().getName()); started.countDown(); @@ -64,7 +69,7 @@ public class TestLocalExecutionEngine { Assertions.assertTrue(started.await(5, TimeUnit.SECONDS), "process should start"); Assertions.assertTrue( - threadName.get() != null && threadName.get().startsWith("local-snapshots-expiring-"), + threadName.get() != null && threadName.get().startsWith("local-" + tag + "-"), "should run in custom pool"); waitForStatus(identifier, ProcessStatus.SUCCESS, 5000); @@ -148,6 +153,7 @@ public class TestLocalExecutionEngine { Map<String, String> properties = new HashMap<>(); properties.put("pool.default.thread-count", "1"); properties.put("pool.snapshots-expiring.thread-count", "1"); + properties.put("pool.orphan-files-cleaning.thread-count", "1"); properties.put("process.status.ttl", ttl); localEngine.open(properties); return localEngine; diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java index eb89c07c0..f906d4b86 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java @@ -1046,7 +1046,7 @@ public class IcebergTableMaintainer implements TableMaintainer { .latestExpiredSeq; // only expire delete files with sequence-number less or equal to expired data file // there may be some dangling delete files, they will be cleaned by - // OrphanFileCleaningExecutor + // OrphanFilesCleaningProcess return fileEntry.getFile().dataSequenceNumber() <= seqUpperBound; } else { return true; diff --git a/charts/amoro/templates/amoro-configmap.yaml b/charts/amoro/templates/amoro-configmap.yaml index e0638059d..d00566f2c 100644 --- a/charts/amoro/templates/amoro-configmap.yaml +++ b/charts/amoro/templates/amoro-configmap.yaml @@ -103,10 +103,6 @@ data: enabled: true thread-count: 10 - clean-orphan-files: - enabled: true - thread-count: 10 - clean-dangling-delete-files: enabled: true thread-count: 10 diff --git a/dist/src/main/amoro-bin/conf/config.yaml b/dist/src/main/amoro-bin/conf/config.yaml index 65446fe2b..e5fc2654a 100644 --- a/dist/src/main/amoro-bin/conf/config.yaml +++ b/dist/src/main/amoro-bin/conf/config.yaml @@ -112,11 +112,6 @@ ams: enabled: true thread-count: 10 - clean-orphan-files: - enabled: true - thread-count: 10 - interval: 1d - clean-dangling-delete-files: enabled: true thread-count: 10 diff --git a/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml b/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml index 5c3199f32..52e062629 100755 --- a/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml @@ -22,3 +22,4 @@ execute-engines: properties: pool.default.thread-count: 10 pool.snapshots-expiring.thread-count: 10 + pool.orphan-files-cleaning.thread-count: 10 diff --git a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml index e1455c2cf..5825fde34 100755 --- a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml @@ -22,3 +22,5 @@ process-factories: properties: expire-snapshots.enabled: "true" expire-snapshots.interval: "1h" + clean-orphan-files.enabled: "true" + clean-orphan-files.interval: "1d" diff --git a/docs/admin-guides/deployment.md b/docs/admin-guides/deployment.md index 57f131533..060923b3c 100644 --- a/docs/admin-guides/deployment.md +++ b/docs/admin-guides/deployment.md @@ -247,6 +247,70 @@ scrape_configs: - targets: ['localhost:9090'] # The host and port that you configured in Amoro plugins configs file. ``` +### Configure process factories + +Process factories manage table maintenance actions like snapshot expiration and orphan file cleaning. +They are configured in `$AMORO_CONF_DIR/plugins/process-factories.yaml`. + +The configuration format is: + +```yaml +process-factories: + - name: # the unified plugin name. + enabled: # if this plugin is enabled, default is true. + priority: # plugin priority for loading order. + properties: # a map defines properties of plugin. +``` + +Currently, the `iceberg` process factory is available for Iceberg, Mixed-Iceberg, and Mixed-Hive formats: + +```yaml +process-factories: + - name: iceberg + enabled: true + priority: 100 + properties: + expire-snapshots.enabled: "true" # enable snapshots expiring + expire-snapshots.interval: "1h" # interval for expiring snapshots + clean-orphan-files.enabled: "true" # enable orphan files cleaning + clean-orphan-files.interval: "1d" # interval for cleaning orphan files +``` + +{{< hint info >}} +Process-level properties control whether an action is registered. Table-level properties (see [Table Configurations](../../user-guides/configurations/)) control whether a specific table executes the action. +{{< /hint >}} + +### Configure execute engines + +Execute engines define how maintenance processes are executed. The `local` engine runs processes in AMS thread pools. +They are configured in `$AMORO_CONF_DIR/plugins/execute-engines.yaml`. + +The configuration format is: + +```yaml +execute-engines: + - name: # the unified plugin name. + enabled: # if this engine is enabled, default is true. + priority: # engine priority for loading order. + properties: # a map defines properties of engine. +``` + +```yaml +execute-engines: + - name: local + enabled: true + priority: 100 + properties: + pool.default.thread-count: 10 # default thread pool size + pool.snapshots-expiring.thread-count: 10 # thread pool for snapshot expiration + pool.orphan-files-cleaning.thread-count: 10 # thread pool for orphan file cleaning + process.status.ttl: 4h # TTL for process status cache +``` + +{{< hint info >}} +Custom pools use the pattern `pool.<name>.thread-count`. Processes select pools via their `tag()` method, falling back to the default pool if no match. +{{< /hint >}} + ### Configure encrypted configuration items For enhanced security, AMS supports encrypted values for sensitive configuration items such as passwords within `config.yaml`. This prevents plaintext passwords and other critical information from being directly exposed in the configuration file. Currently, AMS provides built-in support for base64 decryption, and users can also implement custom decryption algorithms if needed (see [Using Customized Encryption Method for Configurations](../using-customized-encryption-method/)). diff --git a/docs/configuration/ams-config.md b/docs/configuration/ams-config.md index c00841dae..aac5c4c42 100644 --- a/docs/configuration/ams-config.md +++ b/docs/configuration/ams-config.md @@ -52,9 +52,6 @@ table td:last-child, table th:last-child { width: 40%; word-break: break-all; } | clean-dangling-delete-files.enabled | true | Enable dangling delete files cleaning. | | clean-dangling-delete-files.interval | 1 d | Interval for cleaning dangling delete files. | | clean-dangling-delete-files.thread-count | 10 | The number of threads used for dangling delete files cleaning. | -| clean-orphan-files.enabled | true | Enable orphan files cleaning. | -| clean-orphan-files.interval | 1 d | Interval for cleaning orphan files. | -| clean-orphan-files.thread-count | 10 | The number of threads used for orphan files cleaning. | | data-expiration.enabled | true | Enable data expiration | | data-expiration.interval | 1 d | Execute interval for data expiration | | data-expiration.thread-count | 10 | The number of threads used for data expiring |
