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-07-25 15:41

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	36	const char* to_string(CompactionProfileType type) {
26	36	switch (type) {
27	8	case CompactionProfileType::BASE:
28	8	return "base";
29	22	case CompactionProfileType::CUMULATIVE:
30	22	return "cumulative";
31	6	case CompactionProfileType::FULL:
32	6	return "full";
33	0	case CompactionProfileType::BINLOG:
34	0	return "binlog";
35	36	}
36	0	return "unknown";
37	36	}
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	6.39k	CompactionTaskTracker* CompactionTaskTracker::instance() {
66	6.39k	static CompactionTaskTracker s_instance;
67	6.39k	return &s_instance;
68	6.39k	}
69
70	899	void CompactionTaskTracker::register_task(CompactionTaskInfo info) {
71	899	if (!config::enable_compaction_task_tracker) {
72	2	return;
73	2	}
74	897	std::unique_lock wlock(_mutex);
75	897	_active_tasks[info.compaction_id] = std::move(info);
76	897	}
77
78	824	void CompactionTaskTracker::update_to_running(int64_t compaction_id, const RunningStats& stats) {
79	824	if (!config::enable_compaction_task_tracker) {
80	0	return;
81	0	}
82	824	std::unique_lock wlock(_mutex);
83	824	auto it = _active_tasks.find(compaction_id);
84	826	if (it != _active_tasks.end()) {
85	826	auto& task = it->second;
86	826	task.status = CompactionTaskStatus::RUNNING;
87	826	task.start_time_ms = stats.start_time_ms;
88	826	task.is_vertical = stats.is_vertical;
89	826	task.permits = stats.permits;
90	826	}
91	824	}
92
93		void CompactionTaskTracker::update_progress(int64_t compaction_id, int64_t total_groups,
94	934	int64_t completed_groups) {
95	934	if (!config::enable_compaction_task_tracker) {
96	0	return;
97	0	}
98	934	std::unique_lock wlock(_mutex);
99	934	auto it = _active_tasks.find(compaction_id);
100	934	if (it != _active_tasks.end()) {
101	808	auto& task = it->second;
102	808	task.vertical_total_groups = total_groups;
103	808	task.vertical_completed_groups = completed_groups;
104	808	}
105	934	}
106
107	632	void CompactionTaskTracker::complete(int64_t compaction_id, const CompletionStats& stats) {
108	632	if (!config::enable_compaction_task_tracker) {
109	0	return;
110	0	}
111	632	std::unique_lock wlock(_mutex);
112	632	auto it = _active_tasks.find(compaction_id);
113	632	if (it == _active_tasks.end()) {
114	62	LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
115	62	return;
116	62	}
117
118		// Extract the task from active map.
119	570	auto node = _active_tasks.extract(it);
120	570	CompactionTaskInfo& info = node.mapped();
121	570	info.status = CompactionTaskStatus::FINISHED;
122	570	_apply_completion(info, stats);
123
124	570	if (config::compaction_task_tracker_max_records > 0) {
125	570	_completed_tasks.push_back(std::move(info));
126	570	_trim_completed_locked();
127	570	}
128	570	}
129
130		void CompactionTaskTracker::fail(int64_t compaction_id, const CompletionStats& stats,
131	278	const std::string& msg) {
132	278	if (!config::enable_compaction_task_tracker) {
133	0	return;
134	0	}
135	278	std::unique_lock wlock(_mutex);
136	278	auto it = _active_tasks.find(compaction_id);
137	278	if (it == _active_tasks.end()) {
138	2	LOG(WARNING) << "compaction_id " << compaction_id << " not found in active_tasks, skip";
139	2	return;
140	2	}
141
142		// Extract the task from active map.
143	276	auto node = _active_tasks.extract(it);
144	276	CompactionTaskInfo& info = node.mapped();
145	276	info.status = CompactionTaskStatus::FAILED;
146	276	info.status_msg = msg;
147	276	_apply_completion(info, stats);
148
149	276	if (config::compaction_task_tracker_max_records > 0) {
150	276	_completed_tasks.push_back(std::move(info));
151	276	_trim_completed_locked();
152	276	}
153	276	}
154
155	824	void CompactionTaskTracker::remove_task(int64_t compaction_id) {
156	824	if (!config::enable_compaction_task_tracker) {
157	0	return;
158	0	}
159	824	std::unique_lock wlock(_mutex);
160	824	_active_tasks.erase(compaction_id); // idempotent: no-op if already removed
161	824	}
162
163		void CompactionTaskTracker::_apply_completion(CompactionTaskInfo& info,
164	846	const CompletionStats& stats) {
165	846	info.end_time_ms = stats.end_time_ms;
166	846	info.merged_rows = stats.merged_rows;
167	846	info.filtered_rows = stats.filtered_rows;
168	846	info.output_rows = stats.output_rows;
169	846	info.output_row_num = stats.output_row_num;
170	846	info.output_data_size = stats.output_data_size;
171	846	info.output_index_size = stats.output_index_size;
172	846	info.output_total_size = stats.output_total_size;
173	846	info.output_segments_num = stats.output_segments_num;
174	846	info.output_version = stats.output_version;
175	846	info.is_ordered_data_compaction = stats.is_ordered_data_compaction;
176	846	info.merge_latency_ms = stats.merge_latency_ms;
177	846	info.bytes_read_from_local = stats.bytes_read_from_local;
178	846	info.bytes_read_from_remote = stats.bytes_read_from_remote;
179	846	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	846	if (info.input_version_range.empty() && !stats.input_version_range.empty()) {
184	0	info.input_version_range = stats.input_version_range;
185	0	}
186	846	if (info.input_rowsets_count == 0 && stats.input_rowsets_count > 0) {
187	0	info.input_rowsets_count = stats.input_rowsets_count;
188	0	}
189	846	if (info.input_row_num == 0 && stats.input_row_num > 0) {
190	0	info.input_row_num = stats.input_row_num;
191	0	}
192	846	if (info.input_data_size == 0 && stats.input_data_size > 0) {
193	0	info.input_data_size = stats.input_data_size;
194	0	}
195	846	if (info.input_index_size == 0 && stats.input_index_size > 0) {
196	0	info.input_index_size = stats.input_index_size;
197	0	}
198	846	if (info.input_total_size == 0 && stats.input_total_size > 0) {
199	0	info.input_total_size = stats.input_total_size;
200	0	}
201	846	if (info.input_segments_num == 0 && stats.input_segments_num > 0) {
202	0	info.input_segments_num = stats.input_segments_num;
203	0	}
204	846	}
205
206	846	void CompactionTaskTracker::_trim_completed_locked() {
207	846	int32_t max = config::compaction_task_tracker_max_records;
208	846	if (max <= 0) {
209	0	_completed_tasks.clear();
210	0	return;
211	0	}
212	856	while (static_cast<int32_t>(_completed_tasks.size()) > max) {
213	10	_completed_tasks.pop_front();
214	10	}
215	846	}
216
217	856	std::vector<CompactionTaskInfo> CompactionTaskTracker::get_all_tasks() const {
218	856	std::shared_lock rlock(_mutex);
219	856	std::vector<CompactionTaskInfo> result;
220	856	result.reserve(_active_tasks.size() + _completed_tasks.size());
221	4.03k	for (const auto& [id, info] : _active_tasks) {
222	4.03k	result.push_back(info);
223	4.03k	}
224	157k	for (const auto& info : _completed_tasks) {
225	157k	result.push_back(info);
226	157k	}
227	856	return result;
228	856	}
229
230		std::vector<CompactionTaskInfo> CompactionTaskTracker::get_completed_tasks(
231		int64_t tablet_id, int64_t top_n, const std::string& compaction_type,
232	26	int success_filter) const {
233	26	int32_t max = config::compaction_task_tracker_max_records;
234	26	if (max <= 0) {
235	0	return {};
236	0	}
237
238	26	std::shared_lock rlock(_mutex);
239	26	std::vector<CompactionTaskInfo> result;
240	26	int32_t count = 0;
241		// Iterate in reverse order (newest first).
242	134	for (auto it = _completed_tasks.rbegin(); it != _completed_tasks.rend(); ++it) {
243	112	if (count >= max) {
244	0	break;
245	0	}
246	112	count++;
247	112	const auto& record = *it;
248	112	if (tablet_id != 0 && record.tablet_id != tablet_id) {
249	14	continue;
250	14	}
251	98	if (!compaction_type.empty() && compaction_type != to_string(record.compaction_type)) {
252	22	continue;
253	22	}
254	76	if (success_filter == 1 && record.status != CompactionTaskStatus::FINISHED) {
255	2	continue;
256	2	}
257	74	if (success_filter == 0 && record.status != CompactionTaskStatus::FAILED) {
258	8	continue;
259	8	}
260	66	result.push_back(record);
261	66	if (top_n > 0 && static_cast<int64_t>(result.size()) >= top_n) {
262	4	break;
263	4	}
264	66	}
265	26	return result;
266	26	}
267
268	30	void CompactionTaskTracker::clear_for_test() {
269	30	std::unique_lock wlock(_mutex);
270	30	_active_tasks.clear();
271	30	_completed_tasks.clear();
272	30	}
273
274		} // namespace doris