Coverage Report

Created: 2026-10-29 21:44

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 <limits>
28
#include <map>
29
#include <memory>
30
#if defined(__clang__)
31
#pragma clang diagnostic pop
32
#endif
33
#include <optional>
34
#include <string>
35
#include <utility>
36
#include <vector>
37
38
#include "cloud/cloud_tablet.h"
39
#include "common/status.h"
40
#include "io/io_common.h"
41
#include "runtime/runtime_profile.h"
42
#include "storage/compaction_task_tracker.h"
43
#include "storage/merger.h"
44
#include "storage/olap_common.h"
45
#include "storage/rowid_conversion.h"
46
#include "storage/rowset/pending_rowset_helper.h"
47
#include "storage/rowset/rowset_fwd.h"
48
#include "storage/tablet/tablet_fwd.h"
49
50
namespace doris {
51
52
class MemTrackerLimiter;
53
class RowsetWriter;
54
struct RowsetWriterContext;
55
class StorageEngine;
56
class CloudStorageEngine;
57
58
static constexpr int COMPACTION_DELETE_BITMAP_LOCK_ID = -1;
59
static constexpr int64_t INVALID_COMPACTION_INITIATOR_ID = -100;
60
// This class is a base class for compaction.
61
// The entrance of this class is compact()
62
// Any compaction should go through four procedures.
63
//  1. pick rowsets satisfied to compact
64
//  2. do compaction
65
//  3. modify rowsets
66
//  4. gc output rowset if failed
67
class Compaction {
68
public:
69
    Compaction(BaseTabletSPtr tablet, const std::string& label);
70
    virtual ~Compaction();
71
72
    // Pick input rowsets satisfied to compact
73
    virtual Status prepare_compact() = 0;
74
75
    // Merge input rowsets to output rowset and modify tablet meta
76
    virtual Status execute_compact() = 0;
77
78
0
    RuntimeProfile* runtime_profile() const { return _profile.get(); }
79
80
    virtual ReaderType compaction_type() const = 0;
81
    virtual std::string_view compaction_name() const = 0;
82
83
    // Returns compaction profile type for task tracking.
84
    // Default returns std::nullopt (not tracked). Only base/cumulative/full override.
85
0
    virtual std::optional<CompactionProfileType> profile_type() const { return std::nullopt; }
86
87
    // Accessors for tracker registration
88
27
    int64_t compaction_id() const { return _compaction_id; }
89
22
    int64_t input_rowsets_data_size() const { return _input_rowsets_data_size; }
90
22
    int64_t input_rowsets_index_size() const { return _input_rowsets_index_size; }
91
22
    int64_t input_rowsets_total_size() const { return _input_rowsets_total_size; }
92
22
    int64_t input_row_num_value() const { return _input_row_num; }
93
22
    int64_t input_rowsets_count() const { return static_cast<int64_t>(_input_rowsets.size()); }
94
64
    virtual int64_t input_segments_num_value() const { return _input_num_segments; }
95
22
    bool is_vertical() const { return _is_vertical; }
96
    std::string input_version_range_str() const;
97
98
    // the difference between index change compmaction and other compaction.
99
    // 1. delete predicate should be kept when input is cumu rowset.
100
    // 2. inverted compaction should be skipped.
101
    // 3. compute level should not be changed.
102
4
    virtual bool is_index_change_compaction() { return false; }
103
104
private:
105
    void set_delete_predicate_for_output_rowset();
106
107
protected:
108
    struct MergeInputRowsetsResult {
109
        bool is_segment_grouped = false;
110
        int64_t segment_group_size = 0;
111
        std::vector<int32_t> output_segment_group_sizes;
112
    };
113
114
    struct MergeInputRowsetsContext {
115
        MergeInputRowsetsResult result;
116
        std::vector<RowsetReaderSharedPtr> input_rs_readers;
117
    };
118
119
    Status merge_input_rowsets();
120
121
    Status prepare_merge_input_rowsets_execution(MergeInputRowsetsContext* context);
122
123
    Status execute_merge_input_rowsets(MergeInputRowsetsContext* context);
124
125
    Status finish_merge_input_rowsets_execution(MergeInputRowsetsContext* context);
126
    void snapshot_changelog_ttl();
127
    void snapshot_row_ttl();
128
    bool pick_expired_changelog_rowset(const std::vector<RowsetSharedPtr>& candidates);
129
    void filter_changelog_ttl_rowsets();
130
131
45
    virtual Status prepare_merge_input_rowsets(MergeInputRowsetsResult* /*result*/) {
132
45
        return Status::OK();
133
45
    }
134
135
    virtual Status do_merge_input_rowsets(
136
            const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
137
            MergeInputRowsetsResult* result);
138
139
45
    virtual void update_output_rowset_after_build(const MergeInputRowsetsResult& /*result*/) {}
140
141
    // Maps one segment-range merge's column-group progress into the whole compaction task.
142
    struct VerticalMergeProgressContext {
143
        int64_t total_ranges;
144
        int64_t range_index;
145
    };
146
147
    Status execute_merge(const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
148
                         int64_t merge_way_num, Merger::Statistics* stats,
149
                         std::optional<std::pair<int64_t, int64_t>> segment_range = std::nullopt,
150
                         VerticalMergeProgressContext progress = {1, 0});
151
152
    // merge inverted index files
153
    Status do_inverted_index_compaction();
154
155
    // mark all columns in columns_to_do_index_compaction to skip index compaction next time.
156
    void mark_skip_index_compaction(const RowsetWriterContext& context,
157
                                    const std::function<void(int64_t, int64_t)>& error_handler);
158
159
    void construct_index_compaction_columns(RowsetWriterContext& ctx);
160
    // SNII native-merge eligibility per (column unique id, index id). merge_input_rowsets()
161
    // plans index compaction before and after the Variant schema extension; each logical index
162
    // is validated (source readers opened, shapes compared, memory preflighted) exactly once and
163
    // the decision is reused, so the second planning pass only adds indexes the extension exposed.
164
    std::map<std::pair<int32_t, int64_t>, Status> _snii_merge_eligibility;
165
    // Native merges rejected by the memory preflight above (rebuilt from the raw column instead).
166
    size_t _snii_merge_preflight_rejections = 0;
167
168
    virtual Status construct_output_rowset_writer(RowsetWriterContext& ctx) = 0;
169
170
    Status check_correctness();
171
172
    int64_t get_avg_segment_rows();
173
174
    void init_profile(const std::string& label);
175
176
    void _load_segment_to_cache();
177
178
    int64_t merge_way_num();
179
180
    virtual Status update_delete_bitmap() = 0;
181
182
    int64_t _compaction_id {0};
183
184
    void submit_profile_record(bool success, int64_t start_time_ms,
185
                               const std::string& status_msg = "");
186
187
    // the root tracker for this compaction
188
    std::shared_ptr<MemTrackerLimiter> _mem_tracker;
189
190
    BaseTabletSPtr _tablet;
191
192
    std::vector<RowsetSharedPtr> _input_rowsets;
193
    std::vector<RowsetSharedPtr> _data_input_rowsets;
194
    int64_t _changelog_ttl_cutoff_tso {-1};
195
    int64_t _changelog_ttl_seconds {0};
196
    int64_t _changelog_ttl_visible_version {std::numeric_limits<int64_t>::max()};
197
    int64_t _changelog_ttl_filtered_rows {0};
198
    int64_t _input_rowsets_data_size {0};
199
    int64_t _input_rowsets_index_size {0};
200
    int64_t _input_rowsets_total_size {0};
201
    int64_t _input_row_num {0};
202
    int64_t _input_num_segments {0};
203
204
    int64_t _local_read_bytes_total {};
205
    int64_t _remote_read_bytes_total {};
206
207
    int64_t _input_rowsets_cached_data_size {0};
208
    int64_t _input_rowsets_cached_index_size {0};
209
210
    Merger::Statistics _stats;
211
212
    RowsetSharedPtr _output_rowset;
213
    std::unique_ptr<RowsetWriter> _output_rs_writer;
214
215
    enum class CompactionState : uint8_t { INITED = 0, SUCCESS = 1 };
216
    CompactionState _state {CompactionState::INITED};
217
218
    bool _is_vertical;
219
    bool _is_ordered_data_compaction {false};
220
    bool _trigger_quick_merge_by_changelog {false};
221
    bool _allow_delete_in_cumu_compaction;
222
    bool _enable_vertical_compact_variant_subcolumns;
223
224
    Version _output_version;
225
226
    int64_t _newest_write_timestamp {-1};
227
    std::unique_ptr<RowIdConversion> _rowid_conversion = nullptr;
228
    TabletSchemaSPtr _cur_tablet_schema;
229
    // The variant path layout of _cur_tablet_schema; empty unless it is an extended schema.
230
    VariantCompactionPathsSPtr _cur_variant_compaction_paths;
231
232
    std::unique_ptr<RuntimeProfile> _profile;
233
234
    bool _enable_inverted_index_compaction {false};
235
236
    RuntimeProfile::Counter* _input_rowsets_data_size_counter = nullptr;
237
    RuntimeProfile::Counter* _input_rowsets_counter = nullptr;
238
    RuntimeProfile::Counter* _input_row_num_counter = nullptr;
239
    RuntimeProfile::Counter* _input_segments_num_counter = nullptr;
240
    RuntimeProfile::Counter* _merged_rows_counter = nullptr;
241
    RuntimeProfile::Counter* _filtered_rows_counter = nullptr;
242
    RuntimeProfile::Counter* _output_rowset_data_size_counter = nullptr;
243
    RuntimeProfile::Counter* _output_row_num_counter = nullptr;
244
    RuntimeProfile::Counter* _output_segments_num_counter = nullptr;
245
    RuntimeProfile::Counter* _merge_rowsets_latency_timer = nullptr;
246
    RuntimeProfile::Counter* _merge_row_data_latency_timer = nullptr;
247
    RuntimeProfile::Counter* _inverted_index_compaction_latency_timer = nullptr;
248
    RuntimeProfile::Counter* _build_output_rowset_latency_timer = nullptr;
249
};
250
251
// `StorageEngine` mixin for `Compaction`
252
class CompactionMixin : public Compaction {
253
public:
254
    CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet, const std::string& label);
255
256
    ~CompactionMixin() override;
257
258
    Status execute_compact() override;
259
260
    int64_t get_compaction_permits();
261
262
0
    int64_t initiator() const { return INVALID_COMPACTION_INITIATOR_ID; }
263
264
    int64_t calc_input_rowsets_total_size() const;
265
266
    int64_t calc_input_rowsets_row_num() const;
267
268
protected:
269
    // Convert `_tablet` from `BaseTablet` to `Tablet`
270
    Tablet* tablet();
271
272
    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
273
274
    virtual Status modify_rowsets();
275
276
    Status update_delete_bitmap() override;
277
278
    void find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,
279
                                          std::vector<Version>* missing_version);
280
281
    StorageEngine& _engine;
282
283
private:
284
    Status execute_compact_impl(int64_t permits);
285
286
    Status build_basic_info(bool is_ordered_compaction = false);
287
288
    // Return true if do ordered data compaction successfully
289
    bool handle_ordered_data_compaction();
290
291
    Status do_compact_ordered_rowsets();
292
293
    bool _check_if_includes_input_rowsets(const RowsetIdUnorderedSet& commit_rowset_ids_set) const;
294
295
    void update_compaction_level();
296
297
    PendingRowsetGuard _pending_rs_guard;
298
};
299
300
class CloudCompactionMixin : public Compaction {
301
public:
302
    CloudCompactionMixin(CloudStorageEngine& engine, CloudTabletSPtr tablet,
303
                         const std::string& label);
304
305
62
    ~CloudCompactionMixin() override = default;
306
307
    Status execute_compact() override;
308
309
    int64_t initiator() const;
310
311
    int64_t num_input_rowsets() const;
312
313
protected:
314
380
    CloudTablet* cloud_tablet() { return static_cast<CloudTablet*>(_tablet.get()); }
315
316
    Status update_delete_bitmap() override;
317
318
    virtual Status garbage_collection();
319
320
    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
321
322
    // Helper function to apply truncation and log the result
323
    // Returns the number of rowsets that were truncated
324
    size_t apply_txn_size_truncation_and_log(const std::string& compaction_name);
325
326
    // Caller must hold the tablet header lock.
327
    bool should_apply_cumulative_compaction_result(int64_t response_cumulative_compaction_cnt);
328
329
    Status prepare_execute_compact(int64_t permits);
330
331
    Status finish_execute_compact(int64_t execution_start_time_us);
332
333
    int64_t get_compaction_permits();
334
335
    CloudStorageEngine& _engine;
336
337
    std::string _uuid;
338
339
    int64_t _expiration = 0;
340
341
0
    virtual Status rebuild_tablet_schema() { return Status::OK(); }
342
343
private:
344
    Status set_storage_resource_from_input_rowsets(RowsetWriterContext& ctx);
345
346
    Status execute_compact_impl(int64_t permits);
347
348
    Status build_basic_info();
349
350
    virtual Status modify_rowsets();
351
352
    void update_compaction_level();
353
354
    bool should_cache_compaction_output();
355
};
356
357
namespace cloud {
358
359
// Truncate rowsets based on estimated transaction metadata size
360
// Returns the number of rowsets that were truncated (removed from the end)
361
// Only considers rowset meta size (using doris_rowset_meta_to_cloud for estimation)
362
size_t truncate_rowsets_by_txn_size(std::vector<RowsetSharedPtr>& rowsets, int64_t& kept_size_bytes,
363
                                    int64_t& truncated_size_bytes);
364
365
} // namespace cloud
366
367
} // namespace doris