be/src/storage/compaction_task_tracker.cpp

Source
// 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.

#include "storage/compaction_task_tracker.h"

#include "common/config.h"
#include "common/logging.h"

namespace doris {

const char* to_string(CompactionProfileType type) {
    switch (type) {
    case CompactionProfileType::BASE:
        return "base";
    case CompactionProfileType::CUMULATIVE:
        return "cumulative";
    case CompactionProfileType::FULL:
        return "full";
    case CompactionProfileType::BINLOG:
        return "binlog";
    }
    return "unknown";
}

const char* to_string(CompactionTaskStatus status) {
    switch (status) {
    case CompactionTaskStatus::PENDING:
        return "PENDING";
    case CompactionTaskStatus::RUNNING:
        return "RUNNING";
    case CompactionTaskStatus::FINISHED:
        return "FINISHED";
    case CompactionTaskStatus::FAILED:
        return "FAILED";
    }
    return "UNKNOWN";
}

const char* to_string(TriggerMethod method) {
    switch (method) {
    case TriggerMethod::AUTO:
        return "AUTO";
    case TriggerMethod::MANUAL:
        return "MANUAL";
    case TriggerMethod::LOAD_TRIGGERED:
        return "LOAD_TRIGGERED";
    }
    return "UNKNOWN";
}

CompactionTaskTracker* CompactionTaskTracker::instance() {
    static CompactionTaskTracker s_instance;
    return &s_instance;
}

void CompactionTaskTracker::register_task(CompactionTaskInfo info) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    _active_tasks[info.compaction_id] = std::move(info);
}

void CompactionTaskTracker::update_to_running(int64_t compaction_id, const RunningStats& stats) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    auto it = _active_tasks.find(compaction_id);
    if (it != _active_tasks.end()) {
        auto& task = it->second;
        task.status = CompactionTaskStatus::RUNNING;
        task.start_time_ms = stats.start_time_ms;
        task.is_vertical = stats.is_vertical;
        task.permits = stats.permits;
    }
}

void CompactionTaskTracker::update_progress(int64_t compaction_id, int64_t total_groups,
                                            int64_t completed_groups) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    auto it = _active_tasks.find(compaction_id);
    if (it != _active_tasks.end()) {
        auto& task = it->second;
        task.vertical_total_groups = total_groups;
        task.vertical_completed_groups = completed_groups;
    }
}

void CompactionTaskTracker::complete(int64_t compaction_id, const CompletionStats& stats) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    auto it = _active_tasks.find(compaction_id);
    if (it == _active_tasks.end()) {
        LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
        return;
    }

    // Extract the task from active map.
    auto node = _active_tasks.extract(it);
    CompactionTaskInfo& info = node.mapped();
    info.status = CompactionTaskStatus::FINISHED;
    _apply_completion(info, stats);

    if (config::compaction_task_tracker_max_records > 0) {
        _completed_tasks.push_back(std::move(info));
        _trim_completed_locked();
    }
}

void CompactionTaskTracker::fail(int64_t compaction_id, const CompletionStats& stats,
                                 const std::string& msg) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    auto it = _active_tasks.find(compaction_id);
    if (it == _active_tasks.end()) {
        LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
        return;
    }

    // Extract the task from active map.
    auto node = _active_tasks.extract(it);
    CompactionTaskInfo& info = node.mapped();
    info.status = CompactionTaskStatus::FAILED;
    info.status_msg = msg;
    _apply_completion(info, stats);

    if (config::compaction_task_tracker_max_records > 0) {
        _completed_tasks.push_back(std::move(info));
        _trim_completed_locked();
    }
}

void CompactionTaskTracker::remove_task(int64_t compaction_id) {
    if (!config::enable_compaction_task_tracker) {
        return;
    }
    std::unique_lock wlock(_mutex);
    _active_tasks.erase(compaction_id); // idempotent: no-op if already removed
}

void CompactionTaskTracker::_apply_completion(CompactionTaskInfo& info,
                                              const CompletionStats& stats) {
    info.end_time_ms = stats.end_time_ms;
    info.merged_rows = stats.merged_rows;
    info.filtered_rows = stats.filtered_rows;
    info.output_rows = stats.output_rows;
    info.output_row_num = stats.output_row_num;
    info.output_data_size = stats.output_data_size;
    info.output_index_size = stats.output_index_size;
    info.output_total_size = stats.output_total_size;
    info.output_segments_num = stats.output_segments_num;
    info.output_version = stats.output_version;
    info.is_ordered_data_compaction = stats.is_ordered_data_compaction;
    info.merge_latency_ms = stats.merge_latency_ms;
    info.bytes_read_from_local = stats.bytes_read_from_local;
    info.bytes_read_from_remote = stats.bytes_read_from_remote;
    info.peak_memory_bytes = stats.peak_memory_bytes;
    // Backfill input stats if they were 0 at register time.
    // Local compaction populates _input_rowsets_data_size etc. in build_basic_info()
    // which runs inside execute_compact_impl(), after register_task().
    if (info.input_version_range.empty() && !stats.input_version_range.empty()) {
        info.input_version_range = stats.input_version_range;
    }
    if (info.input_rowsets_count == 0 && stats.input_rowsets_count > 0) {
        info.input_rowsets_count = stats.input_rowsets_count;
    }
    if (info.input_row_num == 0 && stats.input_row_num > 0) {
        info.input_row_num = stats.input_row_num;
    }
    if (info.input_data_size == 0 && stats.input_data_size > 0) {
        info.input_data_size = stats.input_data_size;
    }
    if (info.input_index_size == 0 && stats.input_index_size > 0) {
        info.input_index_size = stats.input_index_size;
    }
    if (info.input_total_size == 0 && stats.input_total_size > 0) {
        info.input_total_size = stats.input_total_size;
    }
    if (info.input_segments_num == 0 && stats.input_segments_num > 0) {
        info.input_segments_num = stats.input_segments_num;
    }
}

void CompactionTaskTracker::_trim_completed_locked() {
    int32_t max = config::compaction_task_tracker_max_records;
    if (max <= 0) {
        _completed_tasks.clear();
        return;
    }
    while (static_cast<int32_t>(_completed_tasks.size()) > max) {
        _completed_tasks.pop_front();
    }
}

std::vector<CompactionTaskInfo> CompactionTaskTracker::get_all_tasks() const {
    std::shared_lock rlock(_mutex);
    std::vector<CompactionTaskInfo> result;
    result.reserve(_active_tasks.size() + _completed_tasks.size());
    for (const auto& [id, info] : _active_tasks) {
        result.push_back(info);
    }
    for (const auto& info : _completed_tasks) {
        result.push_back(info);
    }
    return result;
}

std::vector<CompactionTaskInfo> CompactionTaskTracker::get_completed_tasks(
        int64_t tablet_id, int64_t top_n, const std::string& compaction_type,
        int success_filter) const {
    int32_t max = config::compaction_task_tracker_max_records;
    if (max <= 0) {
        return {};
    }

    std::shared_lock rlock(_mutex);
    std::vector<CompactionTaskInfo> result;
    int32_t count = 0;
    // Iterate in reverse order (newest first).
    for (auto it = _completed_tasks.rbegin(); it != _completed_tasks.rend(); ++it) {
        if (count >= max) {
            break;
        }
        count++;
        const auto& record = *it;
        if (tablet_id != 0 && record.tablet_id != tablet_id) {
            continue;
        }
        if (!compaction_type.empty() && compaction_type != to_string(record.compaction_type)) {
            continue;
        }
        if (success_filter == 1 && record.status != CompactionTaskStatus::FINISHED) {
            continue;
        }
        if (success_filter == 0 && record.status != CompactionTaskStatus::FAILED) {
            continue;
        }
        result.push_back(record);
        if (top_n > 0 && static_cast<int64_t>(result.size()) >= top_n) {
            break;
        }
    }
    return result;
}

void CompactionTaskTracker::clear_for_test() {
    std::unique_lock wlock(_mutex);
    _active_tasks.clear();
    _completed_tasks.clear();
}

} // namespace doris

Coverage Report

Created: 2026-06-02 16:32

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "storage/compaction_task_tracker.h"
19
20		#include "common/config.h"
21		#include "common/logging.h"
22
23		namespace doris {
24
25	18	const char* to_string(CompactionProfileType type) {
26	18	switch (type) {
27	4	case CompactionProfileType::BASE:
28	4	return "base";
29	11	case CompactionProfileType::CUMULATIVE:
30	11	return "cumulative";
31	3	case CompactionProfileType::FULL:
32	3	return "full";
33	0	case CompactionProfileType::BINLOG:
34	0	return "binlog";
35	18	}
36	0	return "unknown";
37	18	}
38
39	0	const char* to_string(CompactionTaskStatus status) {
40	0	switch (status) {
41	0	case CompactionTaskStatus::PENDING:
42	0	return "PENDING";
43	0	case CompactionTaskStatus::RUNNING:
44	0	return "RUNNING";
45	0	case CompactionTaskStatus::FINISHED:
46	0	return "FINISHED";
47	0	case CompactionTaskStatus::FAILED:
48	0	return "FAILED";
49	0	}
50	0	return "UNKNOWN";
51	0	}
52
53	0	const char* to_string(TriggerMethod method) {
54	0	switch (method) {
55	0	case TriggerMethod::AUTO:
56	0	return "AUTO";
57	0	case TriggerMethod::MANUAL:
58	0	return "MANUAL";
59	0	case TriggerMethod::LOAD_TRIGGERED:
60	0	return "LOAD_TRIGGERED";
61	0	}
62	0	return "UNKNOWN";
63	0	}
64
65	3.06k	CompactionTaskTracker* CompactionTaskTracker::instance() {
66	3.06k	static CompactionTaskTracker s_instance;
67	3.06k	return &s_instance;
68	3.06k	}
69
70	450	void CompactionTaskTracker::register_task(CompactionTaskInfo info) {
71	450	if (!config::enable_compaction_task_tracker) {
72	1	return;
73	1	}
74	449	std::unique_lock wlock(_mutex);
75	449	_active_tasks[info.compaction_id] = std::move(info);
76	449	}
77
78	413	void CompactionTaskTracker::update_to_running(int64_t compaction_id, const RunningStats& stats) {
79	413	if (!config::enable_compaction_task_tracker) {
80	0	return;
81	0	}
82	413	std::unique_lock wlock(_mutex);
83	413	auto it = _active_tasks.find(compaction_id);
84	413	if (it != _active_tasks.end()) {
85	413	auto& task = it->second;
86	413	task.status = CompactionTaskStatus::RUNNING;
87	413	task.start_time_ms = stats.start_time_ms;
88	413	task.is_vertical = stats.is_vertical;
89	413	task.permits = stats.permits;
90	413	}
91	413	}
92
93		void CompactionTaskTracker::update_progress(int64_t compaction_id, int64_t total_groups,
94	402	int64_t completed_groups) {
95	402	if (!config::enable_compaction_task_tracker) {
96	0	return;
97	0	}
98	402	std::unique_lock wlock(_mutex);
99	402	auto it = _active_tasks.find(compaction_id);
100	404	if (it != _active_tasks.end()) {
101	404	auto& task = it->second;
102	404	task.vertical_total_groups = total_groups;
103	404	task.vertical_completed_groups = completed_groups;
104	404	}
105	402	}
106
107	286	void CompactionTaskTracker::complete(int64_t compaction_id, const CompletionStats& stats) {
108	286	if (!config::enable_compaction_task_tracker) {
109	0	return;
110	0	}
111	286	std::unique_lock wlock(_mutex);
112	286	auto it = _active_tasks.find(compaction_id);
113	286	if (it == _active_tasks.end()) {
114	1	LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
115	1	return;
116	1	}
117
118		// Extract the task from active map.
119	285	auto node = _active_tasks.extract(it);
120	285	CompactionTaskInfo& info = node.mapped();
121	285	info.status = CompactionTaskStatus::FINISHED;
122	285	_apply_completion(info, stats);
123
124	285	if (config::compaction_task_tracker_max_records > 0) {
125	285	_completed_tasks.push_back(std::move(info));
126	285	_trim_completed_locked();
127	285	}
128	285	}
129
130		void CompactionTaskTracker::fail(int64_t compaction_id, const CompletionStats& stats,
131	139	const std::string& msg) {
132	139	if (!config::enable_compaction_task_tracker) {
133	0	return;
134	0	}
135	139	std::unique_lock wlock(_mutex);
136	139	auto it = _active_tasks.find(compaction_id);
137	139	if (it == _active_tasks.end()) {
138	1	LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
139	1	return;
140	1	}
141
142		// Extract the task from active map.
143	138	auto node = _active_tasks.extract(it);
144	138	CompactionTaskInfo& info = node.mapped();
145	138	info.status = CompactionTaskStatus::FAILED;
146	138	info.status_msg = msg;
147	138	_apply_completion(info, stats);
148
149	138	if (config::compaction_task_tracker_max_records > 0) {
150	138	_completed_tasks.push_back(std::move(info));
151	138	_trim_completed_locked();
152	138	}
153	138	}
154
155	412	void CompactionTaskTracker::remove_task(int64_t compaction_id) {
156	412	if (!config::enable_compaction_task_tracker) {
157	0	return;
158	0	}
159	412	std::unique_lock wlock(_mutex);
160	412	_active_tasks.erase(compaction_id); // idempotent: no-op if already removed
161	412	}
162
163		void CompactionTaskTracker::_apply_completion(CompactionTaskInfo& info,
164	423	const CompletionStats& stats) {
165	423	info.end_time_ms = stats.end_time_ms;
166	423	info.merged_rows = stats.merged_rows;
167	423	info.filtered_rows = stats.filtered_rows;
168	423	info.output_rows = stats.output_rows;
169	423	info.output_row_num = stats.output_row_num;
170	423	info.output_data_size = stats.output_data_size;
171	423	info.output_index_size = stats.output_index_size;
172	423	info.output_total_size = stats.output_total_size;
173	423	info.output_segments_num = stats.output_segments_num;
174	423	info.output_version = stats.output_version;
175	423	info.is_ordered_data_compaction = stats.is_ordered_data_compaction;
176	423	info.merge_latency_ms = stats.merge_latency_ms;
177	423	info.bytes_read_from_local = stats.bytes_read_from_local;
178	423	info.bytes_read_from_remote = stats.bytes_read_from_remote;
179	423	info.peak_memory_bytes = stats.peak_memory_bytes;
180		// Backfill input stats if they were 0 at register time.
181		// Local compaction populates _input_rowsets_data_size etc. in build_basic_info()
182		// which runs inside execute_compact_impl(), after register_task().
183	423	if (info.input_version_range.empty() && !stats.input_version_range.empty()) {
184	0	info.input_version_range = stats.input_version_range;
185	0	}
186	423	if (info.input_rowsets_count == 0 && stats.input_rowsets_count > 0) {
187	0	info.input_rowsets_count = stats.input_rowsets_count;
188	0	}
189	423	if (info.input_row_num == 0 && stats.input_row_num > 0) {
190	0	info.input_row_num = stats.input_row_num;
191	0	}
192	423	if (info.input_data_size == 0 && stats.input_data_size > 0) {
193	0	info.input_data_size = stats.input_data_size;
194	0	}
195	423	if (info.input_index_size == 0 && stats.input_index_size > 0) {
196	0	info.input_index_size = stats.input_index_size;
197	0	}
198	423	if (info.input_total_size == 0 && stats.input_total_size > 0) {
199	0	info.input_total_size = stats.input_total_size;
200	0	}
201	423	if (info.input_segments_num == 0 && stats.input_segments_num > 0) {
202	0	info.input_segments_num = stats.input_segments_num;
203	0	}
204	423	}
205
206	423	void CompactionTaskTracker::_trim_completed_locked() {
207	423	int32_t max = config::compaction_task_tracker_max_records;
208	423	if (max <= 0) {
209	0	_completed_tasks.clear();
210	0	return;
211	0	}
212	428	while (static_cast<int32_t>(_completed_tasks.size()) > max) {
213	5	_completed_tasks.pop_front();
214	5	}
215	423	}
216
217	426	std::vector<CompactionTaskInfo> CompactionTaskTracker::get_all_tasks() const {
218	426	std::shared_lock rlock(_mutex);
219	426	std::vector<CompactionTaskInfo> result;
220	426	result.reserve(_active_tasks.size() + _completed_tasks.size());
221	2.47k	for (const auto& [id, info] : _active_tasks) {
222	2.47k	result.push_back(info);
223	2.47k	}
224	72.5k	for (const auto& info : _completed_tasks) {
225	72.5k	result.push_back(info);
226	72.5k	}
227	426	return result;
228	426	}
229
230		std::vector<CompactionTaskInfo> CompactionTaskTracker::get_completed_tasks(
231		int64_t tablet_id, int64_t top_n, const std::string& compaction_type,
232	13	int success_filter) const {
233	13	int32_t max = config::compaction_task_tracker_max_records;
234	13	if (max <= 0) {
235	0	return {};
236	0	}
237
238	13	std::shared_lock rlock(_mutex);
239	13	std::vector<CompactionTaskInfo> result;
240	13	int32_t count = 0;
241		// Iterate in reverse order (newest first).
242	67	for (auto it = _completed_tasks.rbegin(); it != _completed_tasks.rend(); ++it) {
243	56	if (count >= max) {
244	0	break;
245	0	}
246	56	count++;
247	56	const auto& record = *it;
248	56	if (tablet_id != 0 && record.tablet_id != tablet_id) {
249	7	continue;
250	7	}
251	49	if (!compaction_type.empty() && compaction_type != to_string(record.compaction_type)) {
252	11	continue;
253	11	}
254	38	if (success_filter == 1 && record.status != CompactionTaskStatus::FINISHED) {
255	1	continue;
256	1	}
257	37	if (success_filter == 0 && record.status != CompactionTaskStatus::FAILED) {
258	4	continue;
259	4	}
260	33	result.push_back(record);
261	33	if (top_n > 0 && static_cast<int64_t>(result.size()) >= top_n) {
262	2	break;
263	2	}
264	33	}
265	13	return result;
266	13	}
267
268	15	void CompactionTaskTracker::clear_for_test() {
269	15	std::unique_lock wlock(_mutex);
270	15	_active_tasks.clear();
271	15	_completed_tasks.clear();
272	15	}
273
274		} // namespace doris