This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 97b19c76b01 [feat](stats) Limit job record #25576 (#26251)
97b19c76b01 is described below
commit 97b19c76b01bded485fd5e372c1b51096ee21afb
Author: AKIRA <[email protected]>
AuthorDate: Wed Nov 1 23:44:43 2023 +0800
[feat](stats) Limit job record #25576 (#26251)
---
.../src/main/java/org/apache/doris/common/Config.java | 2 +-
.../org/apache/doris/statistics/AnalysisManager.java | 17 +++++++++++++----
2 files changed, 14 insertions(+), 5 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index b5f7b0ce92b..ce90a0c1dda 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2168,7 +2168,7 @@ public class Config extends ConfigBase {
"控制统计信息的自动触发作业执行记录的持久化行数",
"Determine the persist number of automatic triggered analyze job
execution status"
})
- public static long auto_analyze_job_record_count = 20000;
+ public static long analyze_record_limit = 20000;
@ConfField(description = {
"Auto Buckets中最小的buckets数目",
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index 3894bd5935c..98c21310905 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -86,6 +86,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.NavigableMap;
import java.util.Optional;
import java.util.Set;
import java.util.StringJoiner;
@@ -112,10 +113,12 @@ public class AnalysisManager extends Daemon implements
Writable {
private AnalysisTaskExecutor taskExecutor;
// Store task information in metadata.
- private final Map<Long, AnalysisInfo> analysisTaskInfoMap =
Collections.synchronizedMap(new TreeMap<>());
+ private final NavigableMap<Long, AnalysisInfo> analysisTaskInfoMap =
+ Collections.synchronizedNavigableMap(new TreeMap<>());
- // Store job information in metadata
- private final Map<Long, AnalysisInfo> analysisJobInfoMap =
Collections.synchronizedMap(new TreeMap<>());
+ // Store job information in metadata.
+ private final NavigableMap<Long, AnalysisInfo> analysisJobInfoMap =
+ Collections.synchronizedNavigableMap(new TreeMap<>());
// Tracking system submitted job, keep in mem only
protected final Map<Long, AnalysisInfo> systemJobInfoMap = new
ConcurrentHashMap<>();
@@ -818,10 +821,16 @@ public class AnalysisManager extends Daemon implements
Writable {
}
public void replayCreateAnalysisJob(AnalysisInfo jobInfo) {
+ while (analysisJobInfoMap.size() >= Config.analyze_record_limit) {
+
analysisJobInfoMap.remove(analysisJobInfoMap.pollFirstEntry().getKey());
+ }
this.analysisJobInfoMap.put(jobInfo.jobId, jobInfo);
}
public void replayCreateAnalysisTask(AnalysisInfo taskInfo) {
+ while (analysisTaskInfoMap.size() >= Config.analyze_record_limit) {
+
analysisTaskInfoMap.remove(analysisTaskInfoMap.pollFirstEntry().getKey());
+ }
this.analysisTaskInfoMap.put(taskInfo.taskId, taskInfo);
}
@@ -1074,7 +1083,7 @@ public class AnalysisManager extends Daemon implements
Writable {
protected SimpleQueue<AnalysisInfo>
createSimpleQueue(Collection<AnalysisInfo> collection,
AnalysisManager analysisManager) {
- return new SimpleQueue<>(Config.auto_analyze_job_record_count,
+ return new SimpleQueue<>(Config.analyze_record_limit,
a -> {
// FE is not ready when replaying log and operations
triggered by replaying
// shouldn't be logged again.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
