This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 5c8990523 [hotfix] Implement an independent process history cleanup
mechanism and migrate the configuration item to the Duration type. (#4161)
5c8990523 is described below
commit 5c8990523f56f11102fd813f8fa90f90afb99631
Author: ConradJam <[email protected]>
AuthorDate: Tue Apr 7 21:06:28 2026 +0800
[hotfix] Implement an independent process history cleanup mechanism and
migrate the configuration item to the Duration type. (#4161)
---
.../apache/amoro/server/AmoroManagementConf.java | 40 ++++-
.../apache/amoro/server/AmoroServiceContainer.java | 2 +-
.../persistence/mapper/TableProcessMapper.java | 7 +
.../scheduler/inline/InlineTableExecutors.java | 30 +++-
...cutor.java => ProcessDataExpiringExecutor.java} | 56 +++---
.../amoro/server/TestAmoroManagementConf.java | 26 +++
.../inline/TestConfigurableIntervalExecutors.java | 58 ++++++
.../inline/TestProcessDataExpiringExecutor.java | 194 +++++++++++++++++++++
.../src/test/resources/config-with-units.yaml | 5 +
dist/src/main/amoro-bin/conf/config.yaml | 7 +-
docs/configuration/ams-config.md | 8 +-
11 files changed, 397 insertions(+), 36 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 3a31933bf..98331ad33 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
@@ -479,19 +479,37 @@ public class AmoroManagementConf {
.defaultValue(10)
.withDescription("The number of threads that self-optimizing uses to
submit results.");
+ /** @deprecated Use {@link #OPTIMIZING_RUNTIME_DATA_KEEP_TIME} instead. */
+ @Deprecated
public static final ConfigOption<Integer> OPTIMIZING_RUNTIME_DATA_KEEP_DAYS =
ConfigOptions.key("self-optimizing.runtime-data-keep-days")
.intType()
.defaultValue(30)
.withDescription(
- "The number of days that self-optimizing runtime data keeps the
runtime.");
+ "Deprecated: use 'self-optimizing.runtime-data-keep-time'
instead. "
+ + "The number of days that self-optimizing runtime data
keeps the runtime.");
+ /** @deprecated Use {@link #OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL}
instead. */
+ @Deprecated
public static final ConfigOption<Integer>
OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL_HOURS =
ConfigOptions.key("self-optimizing.runtime-data-expire-interval-hours")
.intType()
.defaultValue(1)
.withDescription(
- "The number of hours that self-optimizing runtime data expire
interval.");
+ "Deprecated: use 'self-optimizing.runtime-data-expire-interval'
instead. "
+ + "The number of hours that self-optimizing runtime data
expire interval.");
+
+ public static final ConfigOption<Duration> OPTIMIZING_RUNTIME_DATA_KEEP_TIME
=
+ ConfigOptions.key("self-optimizing.runtime-data-keep-time")
+ .durationType()
+ .defaultValue(Duration.ofDays(30))
+ .withDescription("Duration that self-optimizing runtime data is
retained.");
+
+ public static final ConfigOption<Duration>
OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL =
+ ConfigOptions.key("self-optimizing.runtime-data-expire-interval")
+ .durationType()
+ .defaultValue(Duration.ofHours(1))
+ .withDescription("Interval between self-optimizing runtime data
expiration runs.");
public static final ConfigOption<Boolean>
OPTIMIZING_BREAK_QUOTA_LIMIT_ENABLED =
ConfigOptions.key("self-optimizing.break-quota-limit-enabled")
@@ -500,6 +518,24 @@ public class AmoroManagementConf {
.withDescription(
"Allow the table to break the quota limit when the resource is
sufficient.");
+ /** @deprecated Use {@link #PROCESS_HISTORY_DATA_KEEP_TIME} instead. */
+ @Deprecated
+ public static final ConfigOption<Integer> PROCESS_HISTORY_DATA_KEEP_DAYS =
+ ConfigOptions.key("process.history-data-keep-days")
+ .intType()
+ .defaultValue(7)
+ .withDescription(
+ "Deprecated: use 'process.history-data-keep-time' instead. "
+ + "The number of days that process history data is
retained.");
+
+ public static final ConfigOption<Duration> PROCESS_HISTORY_DATA_KEEP_TIME =
+ ConfigOptions.key("process.history-data-keep-time")
+ .durationType()
+ .defaultValue(Duration.ofDays(7))
+ .withDescription(
+ "Duration that process history data is retained. "
+ + "Expired terminal process records will be deleted
automatically.");
+
public static final ConfigOption<Duration> OVERVIEW_CACHE_REFRESH_INTERVAL =
ConfigOptions.key("overview-cache.refresh-interval")
.durationType()
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 0fa2b917e..768dab89f 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
@@ -293,7 +293,7 @@ public class AmoroServiceContainer {
addHandlerChain(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getOptimizingCommitExecutor());
-
addHandlerChain(InlineTableExecutors.getInstance().getOptimizingExpiringExecutor());
+
addHandlerChain(InlineTableExecutors.getInstance().getProcessDataExpiringExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getBlockerExpiringExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getHiveCommitSyncExecutor());
addHandlerChain(InlineTableExecutors.getInstance().getTableRefreshingExecutor());
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java
index ed6192452..bbf4019c7 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java
@@ -42,6 +42,13 @@ public interface TableProcessMapper {
@Delete("DELETE FROM table_process WHERE process_id <= #{processId} AND
table_id = #{tableId}")
void deleteBefore(@Param("tableId") long tableId, @Param("processId") long
processId);
+ @Delete(
+ "DELETE FROM table_process WHERE process_id <= #{minProcessId} "
+ + "AND table_id = #{tableId} "
+ + "AND status NOT IN ('SUBMITTED', 'RUNNING', 'PENDING',
'CANCELING')")
+ void deleteExpiredProcesses(
+ @Param("tableId") long tableId, @Param("minProcessId") long
minProcessId);
+
@Insert(
"INSERT INTO table_process "
+ "(process_id, table_id, external_process_identifier, status,
process_type, process_stage, execution_engine, retry_number, "
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 17eb72e39..173c0259e 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
@@ -22,6 +22,8 @@ import org.apache.amoro.config.Configurations;
import org.apache.amoro.server.AmoroManagementConf;
import org.apache.amoro.server.table.TableService;
+import java.time.Duration;
+
public class InlineTableExecutors {
private static final InlineTableExecutors instance = new
InlineTableExecutors();
@@ -30,7 +32,7 @@ public class InlineTableExecutors {
private DanglingDeleteFilesCleaningExecutor
danglingDeleteFilesCleaningExecutor;
private BlockerExpiringExecutor blockerExpiringExecutor;
private OptimizingCommitExecutor optimizingCommitExecutor;
- private OptimizingExpiringExecutor optimizingExpiringExecutor;
+ private ProcessDataExpiringExecutor processDataExpiringExecutor;
private HiveCommitSyncExecutor hiveCommitSyncExecutor;
private TagsAutoCreatingExecutor tagsAutoCreatingExecutor;
private DataExpiringExecutor dataExpiringExecutor;
@@ -57,11 +59,23 @@ public class InlineTableExecutors {
this.optimizingCommitExecutor =
new OptimizingCommitExecutor(
tableService,
conf.getInteger(AmoroManagementConf.OPTIMIZING_COMMIT_THREAD_COUNT));
- this.optimizingExpiringExecutor =
- new OptimizingExpiringExecutor(
- tableService,
-
conf.getInteger(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_DAYS),
-
conf.getInteger(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL_HOURS));
+ Duration optimizingKeepTime =
+ conf.contains(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_TIME)
+ ? conf.get(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_TIME)
+ : Duration.ofDays(
+
conf.getInteger(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_DAYS));
+ Duration expireInterval =
+
conf.contains(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL)
+ ?
conf.get(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL)
+ : Duration.ofHours(
+
conf.getInteger(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL_HOURS));
+ Duration processKeepTime =
+ conf.contains(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_TIME)
+ ? conf.get(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_TIME)
+ :
Duration.ofDays(conf.getInteger(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_DAYS));
+ this.processDataExpiringExecutor =
+ new ProcessDataExpiringExecutor(
+ tableService, optimizingKeepTime, expireInterval, processKeepTime);
this.blockerExpiringExecutor = new BlockerExpiringExecutor(tableService);
if (conf.getBoolean(AmoroManagementConf.SYNC_HIVE_TABLES_ENABLED)) {
this.hiveCommitSyncExecutor =
@@ -110,8 +124,8 @@ public class InlineTableExecutors {
return optimizingCommitExecutor;
}
- public OptimizingExpiringExecutor getOptimizingExpiringExecutor() {
- return optimizingExpiringExecutor;
+ public ProcessDataExpiringExecutor getProcessDataExpiringExecutor() {
+ return processDataExpiringExecutor;
}
public HiveCommitSyncExecutor getHiveCommitSyncExecutor() {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OptimizingExpiringExecutor.java
b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/ProcessDataExpiringExecutor.java
similarity index 56%
rename from
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OptimizingExpiringExecutor.java
rename to
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/ProcessDataExpiringExecutor.java
index 329afc8ef..1bc4e9ac7 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OptimizingExpiringExecutor.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/ProcessDataExpiringExecutor.java
@@ -28,22 +28,30 @@ import org.apache.amoro.server.utils.SnowflakeIdGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class OptimizingExpiringExecutor extends PeriodicTableScheduler {
- private static final Logger LOG =
LoggerFactory.getLogger(OptimizingExpiringExecutor.class);
+import java.time.Duration;
+
+public class ProcessDataExpiringExecutor extends PeriodicTableScheduler {
+ private static final Logger LOG =
LoggerFactory.getLogger(ProcessDataExpiringExecutor.class);
private final Persistency persistency = new Persistency();
- private final long keepTime;
- private final long interval;
+ private final long optimizingKeepTimeMs;
+ private final long processKeepTimeMs;
+ private final long expireIntervalMs;
- public OptimizingExpiringExecutor(TableService tableService, int keepDays,
int intervalHours) {
+ public ProcessDataExpiringExecutor(
+ TableService tableService,
+ Duration optimizingKeepTime,
+ Duration expireInterval,
+ Duration processKeepTime) {
super(tableService, 1);
- this.keepTime = keepDays * 24 * 60 * 60 * 1000L;
- this.interval = intervalHours * 60 * 60 * 1000L;
+ this.optimizingKeepTimeMs = optimizingKeepTime.toMillis();
+ this.processKeepTimeMs = processKeepTime.toMillis();
+ this.expireIntervalMs = expireInterval.toMillis();
}
@Override
protected long getNextExecutingTime(TableRuntime tableRuntime) {
- return interval;
+ return expireIntervalMs;
}
@Override
@@ -68,32 +76,38 @@ public class OptimizingExpiringExecutor extends
PeriodicTableScheduler {
private class Persistency extends PersistentBase {
public void doExpiring(TableRuntime tableRuntime) {
- long expireTime = System.currentTimeMillis() - keepTime;
- long minProcessId = SnowflakeIdGenerator.getMinSnowflakeId(expireTime);
+ long tableId = tableRuntime.getTableIdentifier().getId();
+ long now = System.currentTimeMillis();
+
+ // 1. Expire optimizing runtime data (optimizingKeepTimeMs, e.g. 30d)
+ long optimizingMinId = SnowflakeIdGenerator.getMinSnowflakeId(now -
optimizingKeepTimeMs);
doAsTransaction(
() ->
doAs(
TableProcessMapper.class,
- mapper ->
-
mapper.deleteBefore(tableRuntime.getTableIdentifier().getId(), minProcessId)),
+ mapper -> mapper.deleteBefore(tableId, optimizingMinId)),
() ->
doAs(
OptimizingProcessMapper.class,
- mapper ->
- mapper.deleteProcessStateBefore(
- tableRuntime.getTableIdentifier().getId(),
minProcessId)),
+ mapper -> mapper.deleteProcessStateBefore(tableId,
optimizingMinId)),
() ->
doAs(
OptimizingProcessMapper.class,
- mapper ->
- mapper.deleteTaskRuntimesBefore(
- tableRuntime.getTableIdentifier().getId(),
minProcessId)),
+ mapper -> mapper.deleteTaskRuntimesBefore(tableId,
optimizingMinId)),
() ->
doAs(
OptimizingProcessMapper.class,
- mapper ->
- mapper.deleteOptimizingQuotaBefore(
- tableRuntime.getTableIdentifier().getId(),
minProcessId)));
+ mapper -> mapper.deleteOptimizingQuotaBefore(tableId,
optimizingMinId)));
+
+ // 2. Expire process history terminal records (processKeepTimeMs, e.g.
7d)
+ // Only deletes terminal records in the window between
processKeepTime and keepTime,
+ // since records older than keepTime are already removed by step 1.
+ if (processKeepTimeMs < optimizingKeepTimeMs) {
+ long processMinId = SnowflakeIdGenerator.getMinSnowflakeId(now -
processKeepTimeMs);
+ doAs(
+ TableProcessMapper.class,
+ mapper -> mapper.deleteExpiredProcesses(tableId, processMinId));
+ }
}
}
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java
index 7b0fcc9d4..99e8b451a 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java
@@ -77,6 +77,32 @@ public class TestAmoroManagementConf {
assertTimeRelatedConfigs(serviceConfig, expectedConfig);
}
+ @Test
+ void testNewDurationConfigDefaults() {
+ Configurations serviceConfig = new Configurations();
+ Assertions.assertEquals(
+ Duration.ofDays(30),
+
serviceConfig.get(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_TIME));
+ Assertions.assertEquals(
+ Duration.ofHours(1),
+
serviceConfig.get(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL));
+ Assertions.assertEquals(
+ Duration.ofDays(7),
serviceConfig.get(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_TIME));
+ }
+
+ @Test
+ void testDeprecatedIntegerConfigDefaults() {
+ Configurations serviceConfig = new Configurations();
+ Assertions.assertEquals(
+ 30,
serviceConfig.getInteger(AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_KEEP_DAYS));
+ Assertions.assertEquals(
+ 1,
+ serviceConfig.getInteger(
+
AmoroManagementConf.OPTIMIZING_RUNTIME_DATA_EXPIRE_INTERVAL_HOURS));
+ Assertions.assertEquals(
+ 7,
serviceConfig.getInteger(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_DAYS));
+ }
+
@Test
void testParsingDefaultStorageRelatedConfigs() {
Configurations serviceConfig = new Configurations();
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestConfigurableIntervalExecutors.java
b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestConfigurableIntervalExecutors.java
index a3d7b2515..5b50245b6 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestConfigurableIntervalExecutors.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestConfigurableIntervalExecutors.java
@@ -64,4 +64,62 @@ public class TestConfigurableIntervalExecutors {
// 5 hours ago - should not execute
Assert.assertFalse(executor.shouldExecute(now -
Duration.ofHours(5).toMillis()));
}
+
+ @Test
+ public void testSnapshotsExpiringDefaultInterval() {
+ Duration interval = Duration.ofHours(1);
+ SnapshotsExpiringExecutor executor = new SnapshotsExpiringExecutor(null,
1, interval);
+
+ TableRuntime tableRuntime = Mockito.mock(TableRuntime.class);
+ Assert.assertEquals(
+ Duration.ofHours(1).toMillis(),
executor.getNextExecutingTime(tableRuntime));
+ }
+
+ @Test
+ public void testSnapshotsExpiringCustomInterval() {
+ Duration interval = Duration.ofMinutes(30);
+ SnapshotsExpiringExecutor executor = new SnapshotsExpiringExecutor(null,
1, interval);
+
+ TableRuntime tableRuntime = Mockito.mock(TableRuntime.class);
+ Assert.assertEquals(
+ Duration.ofMinutes(30).toMillis(),
executor.getNextExecutingTime(tableRuntime));
+ }
+
+ @Test
+ public void testProcessDataExpiringDefaultInterval() {
+ Duration optimizingKeepTime = Duration.ofDays(30);
+ Duration expireInterval = Duration.ofHours(1);
+ Duration processKeepTime = Duration.ofDays(7);
+ ProcessDataExpiringExecutor executor =
+ new ProcessDataExpiringExecutor(null, optimizingKeepTime,
expireInterval, processKeepTime);
+
+ TableRuntime tableRuntime = Mockito.mock(TableRuntime.class);
+ Assert.assertEquals(
+ Duration.ofHours(1).toMillis(),
executor.getNextExecutingTime(tableRuntime));
+ }
+
+ @Test
+ public void testProcessDataExpiringCustomInterval() {
+ Duration optimizingKeepTime = Duration.ofDays(15);
+ Duration expireInterval = Duration.ofMinutes(30);
+ Duration processKeepTime = Duration.ofDays(3);
+ ProcessDataExpiringExecutor executor =
+ new ProcessDataExpiringExecutor(null, optimizingKeepTime,
expireInterval, processKeepTime);
+
+ TableRuntime tableRuntime = Mockito.mock(TableRuntime.class);
+ Assert.assertEquals(
+ Duration.ofMinutes(30).toMillis(),
executor.getNextExecutingTime(tableRuntime));
+ }
+
+ @Test
+ public void testSnapshotsExpiringShouldExecuteAfterInterval() {
+ Duration interval = Duration.ofHours(2);
+ SnapshotsExpiringExecutor executor = new SnapshotsExpiringExecutor(null,
1, interval);
+
+ long now = System.currentTimeMillis();
+ // 3 hours ago - should execute
+ Assert.assertTrue(executor.shouldExecute(now -
Duration.ofHours(3).toMillis()));
+ // 1 hour ago - should not execute
+ Assert.assertFalse(executor.shouldExecute(now -
Duration.ofHours(1).toMillis()));
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestProcessDataExpiringExecutor.java
b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestProcessDataExpiringExecutor.java
new file mode 100644
index 000000000..bd251b02d
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestProcessDataExpiringExecutor.java
@@ -0,0 +1,194 @@
+/*
+ * 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.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.process.ProcessStatus;
+import org.apache.amoro.server.AMSServiceTestBase;
+import org.apache.amoro.server.persistence.PersistentBase;
+import org.apache.amoro.server.persistence.mapper.TableProcessMapper;
+import org.apache.amoro.server.process.TableProcessMeta;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.TableService;
+import org.apache.amoro.server.utils.SnowflakeIdGenerator;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+
+public class TestProcessDataExpiringExecutor extends AMSServiceTestBase {
+
+ private static final long TABLE_ID = 1L;
+ private static final ServerTableIdentifier TABLE_IDENTIFIER =
+ ServerTableIdentifier.of(
+ TABLE_ID, "test_catalog", "test_db", "test_table_expiring",
TableFormat.MIXED_ICEBERG);
+
+ private final Persistency persistency = new Persistency();
+ private DefaultTableRuntime tableRuntime;
+ private TableService tableService;
+
+ @Before
+ public void mock() {
+ tableRuntime = Mockito.mock(DefaultTableRuntime.class);
+ tableService = Mockito.mock(TableService.class);
+
Mockito.when(tableRuntime.getTableIdentifier()).thenReturn(TABLE_IDENTIFIER);
+ // Clean up any leftover data
+ persistency.cleanAll(TABLE_ID);
+ }
+
+ @Test
+ public void testProcessHistoryExpiringWhenShorterThanKeepTime() {
+ // optimizingKeepTime=30d, processKeepTime=7d
+ Duration optimizingKeepTime = Duration.ofDays(30);
+ Duration expireInterval = Duration.ofHours(1);
+ Duration processKeepTime = Duration.ofDays(7);
+
+ long now = System.currentTimeMillis();
+
+ // Insert a terminal (SUCCESS) process at 10 days ago - should be deleted
by processKeepTime
+ long tenDaysAgo = now - Duration.ofDays(10).toMillis();
+ long processId10d = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 1;
+ persistency.insertProcess(TABLE_ID, processId10d, ProcessStatus.SUCCESS,
tenDaysAgo);
+
+ // Insert a terminal (SUCCESS) process at 3 days ago - should be kept
(within processKeepTime)
+ long threeDaysAgo = now - Duration.ofDays(3).toMillis();
+ long processId3d = SnowflakeIdGenerator.getMinSnowflakeId(threeDaysAgo) +
1;
+ persistency.insertProcess(TABLE_ID, processId3d, ProcessStatus.SUCCESS,
threeDaysAgo);
+
+ Assert.assertEquals(2, persistency.listProcesses(TABLE_ID).size());
+
+ ProcessDataExpiringExecutor executor =
+ new ProcessDataExpiringExecutor(
+ tableService, optimizingKeepTime, expireInterval, processKeepTime);
+ executor.execute(tableRuntime);
+
+ List<TableProcessMeta> remaining = persistency.listProcesses(TABLE_ID);
+ Assert.assertEquals(1, remaining.size());
+ Assert.assertEquals(processId3d, remaining.get(0).getProcessId());
+ }
+
+ @Test
+ public void testProcessHistoryNotExpiringWhenEqualToKeepTime() {
+ // When processKeepTime >= optimizingKeepTime, the extra process cleanup
should not trigger
+ Duration optimizingKeepTime = Duration.ofDays(7);
+ Duration expireInterval = Duration.ofHours(1);
+ Duration processKeepTime = Duration.ofDays(7);
+
+ long now = System.currentTimeMillis();
+
+ // Insert a terminal process at 5 days ago - within both
optimizingKeepTime and processKeepTime
+ long fiveDaysAgo = now - Duration.ofDays(5).toMillis();
+ long processId5d = SnowflakeIdGenerator.getMinSnowflakeId(fiveDaysAgo) + 1;
+ persistency.insertProcess(TABLE_ID, processId5d, ProcessStatus.SUCCESS,
fiveDaysAgo);
+
+ Assert.assertEquals(1, persistency.listProcesses(TABLE_ID).size());
+
+ ProcessDataExpiringExecutor executor =
+ new ProcessDataExpiringExecutor(
+ tableService, optimizingKeepTime, expireInterval, processKeepTime);
+ executor.execute(tableRuntime);
+
+ // The process should still exist - not expired by either mechanism
+ Assert.assertEquals(1, persistency.listProcesses(TABLE_ID).size());
+ }
+
+ @Test
+ public void testDeleteExpiredProcessesSkipsActiveStatuses() {
+ // optimizingKeepTime=30d, processKeepTime=7d
+ Duration optimizingKeepTime = Duration.ofDays(30);
+ Duration expireInterval = Duration.ofHours(1);
+ Duration processKeepTime = Duration.ofDays(7);
+
+ long now = System.currentTimeMillis();
+ long tenDaysAgo = now - Duration.ofDays(10).toMillis();
+
+ // Insert active-status processes at 10 days ago - should NOT be deleted
+ long pidRunning = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 1;
+ persistency.insertProcess(TABLE_ID, pidRunning, ProcessStatus.RUNNING,
tenDaysAgo);
+
+ long pidSubmitted = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 2;
+ persistency.insertProcess(TABLE_ID, pidSubmitted, ProcessStatus.SUBMITTED,
tenDaysAgo);
+
+ long pidPending = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 3;
+ persistency.insertProcess(TABLE_ID, pidPending, ProcessStatus.PENDING,
tenDaysAgo);
+
+ long pidCanceling = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 4;
+ persistency.insertProcess(TABLE_ID, pidCanceling, ProcessStatus.CANCELING,
tenDaysAgo);
+
+ // Insert a terminal process at 10 days ago - SHOULD be deleted
+ long pidSuccess = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 5;
+ persistency.insertProcess(TABLE_ID, pidSuccess, ProcessStatus.SUCCESS,
tenDaysAgo);
+
+ long pidFailed = SnowflakeIdGenerator.getMinSnowflakeId(tenDaysAgo) + 6;
+ persistency.insertProcess(TABLE_ID, pidFailed, ProcessStatus.FAILED,
tenDaysAgo);
+
+ Assert.assertEquals(6, persistency.listProcesses(TABLE_ID).size());
+
+ ProcessDataExpiringExecutor executor =
+ new ProcessDataExpiringExecutor(
+ tableService, optimizingKeepTime, expireInterval, processKeepTime);
+ executor.execute(tableRuntime);
+
+ List<TableProcessMeta> remaining = persistency.listProcesses(TABLE_ID);
+ // Active statuses (RUNNING, SUBMITTED, PENDING, CANCELING) should survive
+ Assert.assertEquals(4, remaining.size());
+ for (TableProcessMeta meta : remaining) {
+ Assert.assertTrue(
+ "Expected active status but got: " + meta.getStatus(),
+ meta.getStatus() == ProcessStatus.RUNNING
+ || meta.getStatus() == ProcessStatus.SUBMITTED
+ || meta.getStatus() == ProcessStatus.PENDING
+ || meta.getStatus() == ProcessStatus.CANCELING);
+ }
+ }
+
+ private static class Persistency extends PersistentBase {
+
+ public void insertProcess(long tableId, long processId, ProcessStatus
status, long createTime) {
+ doAs(
+ TableProcessMapper.class,
+ mapper ->
+ mapper.insertProcess(
+ tableId,
+ processId,
+ "",
+ status,
+ "TEST",
+ status.name(),
+ "LOCAL",
+ 0,
+ createTime,
+ Collections.emptyMap(),
+ Collections.emptyMap()));
+ }
+
+ public List<TableProcessMeta> listProcesses(long tableId) {
+ return getAs(TableProcessMapper.class, mapper ->
mapper.listProcessMeta(tableId, null, null));
+ }
+
+ public void cleanAll(long tableId) {
+ doAs(TableProcessMapper.class, mapper -> mapper.deleteBefore(tableId,
Long.MAX_VALUE));
+ }
+ }
+}
diff --git a/amoro-ams/src/test/resources/config-with-units.yaml
b/amoro-ams/src/test/resources/config-with-units.yaml
index 9f72540a9..7f307de7d 100644
--- a/amoro-ams/src/test/resources/config-with-units.yaml
+++ b/amoro-ams/src/test/resources/config-with-units.yaml
@@ -46,10 +46,15 @@ ams:
interval: 1min
max-pending-partition-count: 100 # default 100
+ process:
+ history-data-keep-time: 7d
+
self-optimizing:
commit-thread-count: 10
runtime-data-keep-days: 30
runtime-data-expire-interval-hours: 1
+ runtime-data-keep-time: 30d
+ runtime-data-expire-interval: 1h
refresh-group-interval: 30s
break-quota-limit-enabled: true
diff --git a/dist/src/main/amoro-bin/conf/config.yaml
b/dist/src/main/amoro-bin/conf/config.yaml
index 27d3e5eb4..65446fe2b 100644
--- a/dist/src/main/amoro-bin/conf/config.yaml
+++ b/dist/src/main/amoro-bin/conf/config.yaml
@@ -81,10 +81,13 @@ ams:
interval: 1min # 60000
max-pending-partition-count: 100 # default 100
+ process:
+ history-data-keep-time: 7d
+
self-optimizing:
commit-thread-count: 10
- runtime-data-keep-days: 30
- runtime-data-expire-interval-hours: 1
+ runtime-data-keep-time: 30d
+ runtime-data-expire-interval: 1h
refresh-group-interval: 30s
break-quota-limit-enabled: true
diff --git a/docs/configuration/ams-config.md b/docs/configuration/ams-config.md
index c929ef2a8..3285e57e6 100644
--- a/docs/configuration/ams-config.md
+++ b/docs/configuration/ams-config.md
@@ -111,6 +111,8 @@ table td:last-child, table th:last-child { width: 40%;
word-break: break-all; }
| optimizer.task-execute-timeout | 2147483647 s | Timeout duration for task
execution, default to Integer.MAX_VALUE seconds(about 24,855 days). |
| overview-cache.max-size | 3360 | Max size of overview cache. |
| overview-cache.refresh-interval | 3 min | Interval for refreshing overview
cache. |
+| process.history-data-keep-days | 7 | Deprecated: use
'process.history-data-keep-time' instead. The number of days that process
history data is retained. |
+| process.history-data-keep-time | 7 d | Duration that process history data is
retained. Expired terminal process records will be deleted automatically. |
| refresh-external-catalogs.interval | 3 min | Interval to refresh the
external catalog. |
| refresh-external-catalogs.queue-size | 1000000 | The queue size of the
executors of the external catalog explorer. |
| refresh-external-catalogs.thread-count | 10 | The number of threads used for
discovering tables in external catalogs. |
@@ -122,8 +124,10 @@ table td:last-child, table th:last-child { width: 40%;
word-break: break-all; }
| self-optimizing.commit-thread-count | 10 | The number of threads that
self-optimizing uses to submit results. |
| self-optimizing.plan-manifest-io-thread-count | 10 | Sets the size of the
worker pool. The worker pool limits the number of tasks concurrently processing
manifests in the base table implementation across all concurrent planning
operations. |
| self-optimizing.refresh-group-interval | 30 s | Optimizer group refresh
interval. |
-| self-optimizing.runtime-data-expire-interval-hours | 1 | The number of hours
that self-optimizing runtime data expire interval. |
-| self-optimizing.runtime-data-keep-days | 30 | The number of days that
self-optimizing runtime data keeps the runtime. |
+| self-optimizing.runtime-data-expire-interval | 1 h | Interval between
self-optimizing runtime data expiration runs. |
+| self-optimizing.runtime-data-expire-interval-hours | 1 | Deprecated: use
'self-optimizing.runtime-data-expire-interval' instead. The number of hours
that self-optimizing runtime data expire interval. |
+| self-optimizing.runtime-data-keep-days | 30 | Deprecated: use
'self-optimizing.runtime-data-keep-time' instead. The number of days that
self-optimizing runtime data keeps the runtime. |
+| self-optimizing.runtime-data-keep-time | 30 d | Duration that
self-optimizing runtime data is retained. |
| server-bind-host | 0.0.0.0 | The host bound to the server. |
| server-expose-host | | The exposed host of the server. |
| sync-hive-tables.enabled | false | Enable synchronizing Hive tables. |
