Coverage Report

Created: 2026-05-14 03:59

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/compaction/compaction.h
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
#pragma once
19
20
#include <cstdint>
21
// clang20 + -O1 causes warnings about pass-failed
22
// error: loop not unrolled: the optimizer was unable to perform the requested transformation; the transformation might be disabled or specified as part of an unsupported transformation ordering [-Werror,-Wpass-failed=transform-warning]
23
#if defined(__clang__)
24
#pragma clang diagnostic push
25
#pragma clang diagnostic ignored "-Wpass-failed"
26
#endif
27
#include <memory>
28
#if defined(__clang__)
29
#pragma clang diagnostic pop
30
#endif
31
#include <optional>
32
#include <string>
33
#include <vector>
34
35
#include "cloud/cloud_tablet.h"
36
#include "common/status.h"
37
#include "io/io_common.h"
38
#include "runtime/runtime_profile.h"
39
#include "storage/compaction_task_tracker.h"
40
#include "storage/merger.h"
41
#include "storage/olap_common.h"
42
#include "storage/rowid_conversion.h"
43
#include "storage/rowset/pending_rowset_helper.h"
44
#include "storage/rowset/rowset_fwd.h"
45
#include "storage/segment/variant/variant_doc_path_stats.h"
46
#include "storage/tablet/tablet_fwd.h"
47
48
namespace doris {
49
50
class MemTrackerLimiter;
51
class RowsetWriter;
52
struct RowsetWriterContext;
53
class StorageEngine;
54
class CloudStorageEngine;
55
56
static constexpr int COMPACTION_DELETE_BITMAP_LOCK_ID = -1;
57
static constexpr int64_t INVALID_COMPACTION_INITIATOR_ID = -100;
58
// This class is a base class for compaction.
59
// The entrance of this class is compact()
60
// Any compaction should go through four procedures.
61
//  1. pick rowsets satisfied to compact
62
//  2. do compaction
63
//  3. modify rowsets
64
//  4. gc output rowset if failed
65
class Compaction {
66
public:
67
    Compaction(BaseTabletSPtr tablet, const std::string& label);
68
    virtual ~Compaction();
69
70
    // Pick input rowsets satisfied to compact
71
    virtual Status prepare_compact() = 0;
72
73
    // Merge input rowsets to output rowset and modify tablet meta
74
    virtual Status execute_compact() = 0;
75
76
0
    RuntimeProfile* runtime_profile() const { return _profile.get(); }
77
78
    virtual ReaderType compaction_type() const = 0;
79
    virtual std::string_view compaction_name() const = 0;
80
81
    // Returns compaction profile type for task tracking.
82
    // Default returns std::nullopt (not tracked). Only base/cumulative/full override.
83
824
    virtual std::optional<CompactionProfileType> profile_type() const { return std::nullopt; }
84
85
    // Accessors for tracker registration
86
8.10k
    int64_t compaction_id() const { return _compaction_id; }
87
8.10k
    int64_t input_rowsets_data_size() const { return _input_rowsets_data_size; }
88
8.10k
    int64_t input_rowsets_index_size() const { return _input_rowsets_index_size; }
89
8.10k
    int64_t input_rowsets_total_size() const { return _input_rowsets_total_size; }
90
8.10k
    int64_t input_row_num_value() const { return _input_row_num; }
91
8.10k
    int64_t input_rowsets_count() const { return static_cast<int64_t>(_input_rowsets.size()); }
92
2.87k
    virtual int64_t input_segments_num_value() const { return _input_num_segments; }
93
8.10k
    bool is_vertical() const { return _is_vertical; }
94
    std::string input_version_range_str() const;
95
96
    // the difference between index change compmaction and other compaction.
97
    // 1. delete predicate should be kept when input is cumu rowset.
98
    // 2. inverted compaction should be skipped.
99
    // 3. compute level should not be changed.
100
15.5k
    virtual bool is_index_change_compaction() { return false; }
101
102
private:
103
    void set_delete_predicate_for_output_rowset();
104
105
protected:
106
    Status merge_input_rowsets();
107
108
    // merge inverted index files
109
    Status do_inverted_index_compaction();
110
111
    // mark all columns in columns_to_do_index_compaction to skip index compaction next time.
112
    void mark_skip_index_compaction(const RowsetWriterContext& context,
113
                                    const std::function<void(int64_t, int64_t)>& error_handler);
114
115
    void construct_index_compaction_columns(RowsetWriterContext& ctx);
116
117
    virtual Status construct_output_rowset_writer(RowsetWriterContext& ctx) = 0;
118
119
    Status check_correctness();
120
121
    int64_t get_avg_segment_rows();
122
123
    void init_profile(const std::string& label);
124
125
    void _load_segment_to_cache();
126
127
    int64_t merge_way_num();
128
129
    virtual Status update_delete_bitmap() = 0;
130
131
    int64_t _compaction_id {0};
132
133
    void submit_profile_record(bool success, int64_t start_time_ms,
134
                               const std::string& status_msg = "");
135
136
    // the root tracker for this compaction
137
    std::shared_ptr<MemTrackerLimiter> _mem_tracker;
138
139
    BaseTabletSPtr _tablet;
140
141
    std::vector<RowsetSharedPtr> _input_rowsets;
142
    int64_t _input_rowsets_data_size {0};
143
    int64_t _input_rowsets_index_size {0};
144
    int64_t _input_rowsets_total_size {0};
145
    int64_t _input_row_num {0};
146
    int64_t _input_num_segments {0};
147
148
    int64_t _local_read_bytes_total {};
149
    int64_t _remote_read_bytes_total {};
150
151
    int64_t _input_rowsets_cached_data_size {0};
152
    int64_t _input_rowsets_cached_index_size {0};
153
154
    Merger::Statistics _stats;
155
156
    RowsetSharedPtr _output_rowset;
157
    std::unique_ptr<RowsetWriter> _output_rs_writer;
158
159
    enum CompactionState : uint8_t { INITED = 0, SUCCESS = 1 };
160
    CompactionState _state {CompactionState::INITED};
161
162
    bool _is_vertical;
163
    bool _allow_delete_in_cumu_compaction;
164
    bool _enable_vertical_compact_variant_subcolumns;
165
166
    Version _output_version;
167
168
    int64_t _newest_write_timestamp {-1};
169
    std::unique_ptr<RowIdConversion> _rowid_conversion = nullptr;
170
    TabletSchemaSPtr _cur_tablet_schema;
171
172
    // Per-uid bucket-partitioned path stats collected from source segment
173
    // doc-value stats during build_basic_info. Plumbed into the output
174
    // RowsetWriterContext so VariantDocCompactWriter can pre-reserve its
175
    // sparse-subcolumn accumulators in pure doc-mode compaction.
176
    std::unordered_map<int32_t, std::shared_ptr<segment_v2::VariantDocPathStats>>
177
            _variant_doc_path_stats;
178
179
    std::unique_ptr<RuntimeProfile> _profile;
180
181
    bool _enable_inverted_index_compaction {false};
182
183
    RuntimeProfile::Counter* _input_rowsets_data_size_counter = nullptr;
184
    RuntimeProfile::Counter* _input_rowsets_counter = nullptr;
185
    RuntimeProfile::Counter* _input_row_num_counter = nullptr;
186
    RuntimeProfile::Counter* _input_segments_num_counter = nullptr;
187
    RuntimeProfile::Counter* _merged_rows_counter = nullptr;
188
    RuntimeProfile::Counter* _filtered_rows_counter = nullptr;
189
    RuntimeProfile::Counter* _output_rowset_data_size_counter = nullptr;
190
    RuntimeProfile::Counter* _output_row_num_counter = nullptr;
191
    RuntimeProfile::Counter* _output_segments_num_counter = nullptr;
192
    RuntimeProfile::Counter* _merge_rowsets_latency_timer = nullptr;
193
};
194
195
// `StorageEngine` mixin for `Compaction`
196
class CompactionMixin : public Compaction {
197
public:
198
    CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet, const std::string& label);
199
200
    ~CompactionMixin() override;
201
202
    Status execute_compact() override;
203
204
    int64_t get_compaction_permits();
205
206
0
    int64_t initiator() const { return INVALID_COMPACTION_INITIATOR_ID; }
207
208
    int64_t calc_input_rowsets_total_size() const;
209
210
    int64_t calc_input_rowsets_row_num() const;
211
212
protected:
213
    // Convert `_tablet` from `BaseTablet` to `Tablet`
214
    Tablet* tablet();
215
216
    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
217
218
    virtual Status modify_rowsets();
219
220
    Status update_delete_bitmap() override;
221
222
    StorageEngine& _engine;
223
224
private:
225
    Status execute_compact_impl(int64_t permits);
226
227
    Status build_basic_info(bool is_ordered_compaction = false);
228
229
    // Return true if do ordered data compaction successfully
230
    bool handle_ordered_data_compaction();
231
232
    Status do_compact_ordered_rowsets();
233
234
    bool _check_if_includes_input_rowsets(const RowsetIdUnorderedSet& commit_rowset_ids_set) const;
235
236
    void update_compaction_level();
237
238
    PendingRowsetGuard _pending_rs_guard;
239
};
240
241
class CloudCompactionMixin : public Compaction {
242
public:
243
    CloudCompactionMixin(CloudStorageEngine& engine, CloudTabletSPtr tablet,
244
                         const std::string& label);
245
246
109k
    ~CloudCompactionMixin() override = default;
247
248
    Status execute_compact() override;
249
250
    int64_t initiator() const;
251
252
    int64_t num_input_rowsets() const;
253
254
protected:
255
1.37M
    CloudTablet* cloud_tablet() { return static_cast<CloudTablet*>(_tablet.get()); }
256
257
    Status update_delete_bitmap() override;
258
259
    virtual Status garbage_collection();
260
261
    // Helper function to apply truncation and log the result
262
    // Returns the number of rowsets that were truncated
263
    size_t apply_txn_size_truncation_and_log(const std::string& compaction_name);
264
265
    CloudStorageEngine& _engine;
266
267
    std::string _uuid;
268
269
    int64_t _expiration = 0;
270
271
0
    virtual Status rebuild_tablet_schema() { return Status::OK(); }
272
273
private:
274
    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
275
276
    Status set_storage_resource_from_input_rowsets(RowsetWriterContext& ctx);
277
278
    Status execute_compact_impl(int64_t permits);
279
280
    Status build_basic_info();
281
282
    virtual Status modify_rowsets();
283
284
    int64_t get_compaction_permits();
285
286
    void update_compaction_level();
287
288
    bool should_cache_compaction_output();
289
};
290
291
namespace cloud {
292
293
// Truncate rowsets based on estimated transaction metadata size
294
// Returns the number of rowsets that were truncated (removed from the end)
295
// Only considers rowset meta size (using doris_rowset_meta_to_cloud for estimation)
296
size_t truncate_rowsets_by_txn_size(std::vector<RowsetSharedPtr>& rowsets, int64_t& kept_size_bytes,
297
                                    int64_t& truncated_size_bytes);
298
299
} // namespace cloud
300
301
} // namespace doris