Coverage Report

Created: 2026-08-03 13:02

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/rowset/rowset.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 <butil/macros.h>
21
#include <fmt/format.h>
22
#include <gen_cpp/olap_file.pb.h>
23
#include <gen_cpp/types.pb.h>
24
#include <stddef.h>
25
#include <stdint.h>
26
27
#include <atomic>
28
#include <iterator>
29
#include <memory>
30
#include <mutex>
31
#include <ostream>
32
#include <string>
33
#include <tuple>
34
#include <vector>
35
36
#include "common/cast_set.h"
37
#include "common/logging.h"
38
#include "common/status.h"
39
#include "io/cache/file_cache_common.h"
40
#include "storage/metadata_adder.h"
41
#include "storage/olap_common.h"
42
#include "storage/rowset/rowset_meta.h"
43
#include "storage/tablet/tablet_schema.h"
44
45
namespace doris {
46
47
class Rowset;
48
class RowsetSegmentRange;
49
class RowsetSegmentView;
50
struct RowLocation;
51
52
namespace io {
53
class RemoteFileSystem;
54
} // namespace io
55
56
using RowsetSharedPtr = std::shared_ptr<Rowset>;
57
class RowsetReader;
58
59
// the rowset state transfer graph:
60
//    ROWSET_UNLOADED    <--|
61
//          ↓               |
62
//    ROWSET_LOADED         |
63
//          ↓               |
64
//    ROWSET_UNLOADING   -->|
65
enum RowsetState {
66
    // state for new created rowset
67
    ROWSET_UNLOADED,
68
    // state after load() called
69
    ROWSET_LOADED,
70
    // state for closed() called but owned by some readers
71
    ROWSET_UNLOADING
72
};
73
74
class RowsetStateMachine {
75
public:
76
1.05M
    RowsetStateMachine() : _rowset_state(ROWSET_UNLOADED) {}
77
78
351k
    Status on_load() {
79
351k
        switch (_rowset_state) {
80
351k
        case ROWSET_UNLOADED:
81
351k
            _rowset_state = ROWSET_LOADED;
82
351k
            break;
83
84
0
        default:
85
0
            return Status::Error<ErrorCode::ROWSET_INVALID_STATE_TRANSITION>(
86
0
                    "RowsetStateMachine meet invalid state");
87
351k
        }
88
351k
        return Status::OK();
89
351k
    }
90
91
37.2k
    Status on_close(uint64_t refs_by_reader) {
92
37.2k
        switch (_rowset_state) {
93
37.2k
        case ROWSET_LOADED:
94
37.2k
            if (refs_by_reader == 0) {
95
37.2k
                _rowset_state = ROWSET_UNLOADED;
96
37.2k
            } else {
97
0
                _rowset_state = ROWSET_UNLOADING;
98
0
            }
99
37.2k
            break;
100
101
0
        default:
102
0
            return Status::Error<ErrorCode::ROWSET_INVALID_STATE_TRANSITION>(
103
0
                    "RowsetStateMachine meet invalid state");
104
37.2k
        }
105
37.2k
        return Status::OK();
106
37.2k
    }
107
108
0
    Status on_release() {
109
0
        switch (_rowset_state) {
110
0
        case ROWSET_UNLOADING:
111
0
            _rowset_state = ROWSET_UNLOADED;
112
0
            break;
113
114
0
        default:
115
0
            return Status::Error<ErrorCode::ROWSET_INVALID_STATE_TRANSITION>(
116
0
                    "RowsetStateMachine meet invalid state");
117
0
        }
118
0
        return Status::OK();
119
0
    }
120
121
13.6M
    RowsetState rowset_state() { return _rowset_state; }
122
123
private:
124
    RowsetState _rowset_state;
125
};
126
127
class Rowset : public std::enable_shared_from_this<Rowset>, public MetadataAdder<Rowset> {
128
public:
129
    // Open all segment files in this rowset and load necessary metadata.
130
    // - `use_cache` : whether to use fd cache, only applicable to alpha rowset now
131
    //
132
    // May be called multiple times, subsequent calls will no-op.
133
    // Derived class implements the load logic by overriding the `do_load_once()` method.
134
    Status load(bool use_cache = true);
135
136
    // returns Status::Error<ErrorCode::ROWSET_CREATE_READER>() when failed to create reader
137
    virtual Status create_reader(std::shared_ptr<RowsetReader>* result) = 0;
138
139
166M
    const RowsetMetaSharedPtr& rowset_meta() const { return _rowset_meta; }
140
141
    void merge_rowset_meta(const RowsetMeta& other);
142
143
340
    bool is_pending() const { return _is_pending; }
144
145
21.9M
    bool is_local() const { return _rowset_meta->is_local(); }
146
147
756
    const std::string& tablet_path() const { return _tablet_path; }
148
149
    // publish rowset to make it visible to read
150
    void make_visible(Version version, int64_t commit_tso);
151
    void set_version(Version version);
152
    const TabletSchemaSPtr& tablet_schema() const;
153
154
    // helper class to access RowsetMeta
155
8.89M
    int64_t start_version() const { return rowset_meta()->version().first; }
156
15.5M
    int64_t end_version() const { return rowset_meta()->version().second; }
157
240k
    int64_t index_disk_size() const { return rowset_meta()->index_disk_size(); }
158
8.27M
    int64_t data_disk_size() const { return rowset_meta()->data_disk_size(); }
159
514k
    int64_t total_disk_size() const { return rowset_meta()->total_disk_size(); }
160
9.44k
    bool empty() const { return rowset_meta()->empty(); }
161
0
    bool zero_num_rows() const { return rowset_meta()->num_rows() == 0; }
162
10.5M
    size_t num_rows() const { return rowset_meta()->num_rows(); }
163
22.1M
    Version version() const { return rowset_meta()->version(); }
164
36.2M
    RowsetId rowset_id() const { return rowset_meta()->rowset_id(); }
165
145k
    int64_t creation_time() const { return rowset_meta()->creation_time(); }
166
0
    PUniqueId load_id() const { return rowset_meta()->load_id(); }
167
169k
    int64_t txn_id() const { return rowset_meta()->txn_id(); }
168
0
    int64_t partition_id() const { return rowset_meta()->partition_id(); }
169
    // flag for push delete rowset
170
0
    bool delete_flag() const { return rowset_meta()->delete_flag(); }
171
18.7M
    MOCK_FUNCTION int64_t num_segments() const { return rowset_meta()->num_segments(); }
172
    void to_rowset_pb(RowsetMetaPB* rs_meta) const { return rowset_meta()->to_rowset_pb(rs_meta); }
173
0
    RowsetMetaPB get_rowset_pb() const { return rowset_meta()->get_rowset_pb(); }
174
    // The writing time of the newest data in rowset, to measure the freshness of a rowset.
175
354k
    int64_t newest_write_timestamp() const { return rowset_meta()->newest_write_timestamp(); }
176
    // The commit tso range of the data in rowset.
177
252k
    TsoRange commit_tso() const { return rowset_meta()->commit_tso(); }
178
179
2.02M
    bool is_segments_overlapping() const { return rowset_meta()->is_segments_overlapping(); }
180
4.05M
    KeysType keys_type() { return _schema->keys_type(); }
181
3.23k
    RowsetStatePB rowset_meta_state() const { return rowset_meta()->rowset_state(); }
182
53
    bool produced_by_compaction() const { return rowset_meta()->produced_by_compaction(); }
183
184
    // remove all files in this rowset
185
    // TODO should we rename the method to remove_files() to be more specific?
186
    virtual Status remove() = 0;
187
188
    // close to clear the resource owned by rowset
189
    // including: open files, indexes and so on
190
    // NOTICE: can not call this function in multithreads
191
40.3k
    void close() {
192
40.3k
        RowsetState old_state = _rowset_state_machine.rowset_state();
193
40.3k
        if (old_state != ROWSET_LOADED) {
194
3.09k
            return;
195
3.09k
        }
196
37.2k
        Status st = Status::OK();
197
37.2k
        {
198
37.2k
            std::lock_guard close_lock(_lock);
199
37.2k
            uint64_t current_refs = _refs_by_reader;
200
37.2k
            old_state = _rowset_state_machine.rowset_state();
201
37.2k
            if (old_state != ROWSET_LOADED) {
202
0
                return;
203
0
            }
204
37.2k
            if (current_refs == 0) {
205
37.2k
                do_close();
206
37.2k
            }
207
37.2k
            st = _rowset_state_machine.on_close(current_refs);
208
37.2k
        }
209
37.2k
        if (!st.ok()) {
210
0
            LOG(WARNING) << "state transition failed from:" << _rowset_state_machine.rowset_state();
211
0
            return;
212
0
        }
213
37.2k
        VLOG_NOTICE << "rowset is close. rowset state from:" << old_state << " to "
214
3
                    << _rowset_state_machine.rowset_state() << ", version:" << start_version()
215
3
                    << "-" << end_version() << ", tabletid:" << _rowset_meta->tablet_id();
216
37.2k
    }
217
218
    // hard link all files in this rowset to `dir` to form a new rowset with id `new_rowset_id`.
219
    virtual Status link_files_to(const std::string& dir, RowsetId new_rowset_id,
220
                                 size_t new_rowset_start_seg_id = 0,
221
                                 std::set<int64_t>* without_index_uids = nullptr) = 0;
222
223
    virtual Status get_inverted_index_size(int64_t* index_size) = 0;
224
225
    // copy all files to `dir`
226
    virtual Status copy_files_to(const std::string& dir, const RowsetId& new_rowset_id) = 0;
227
228
    virtual Status upload_to(const StorageResource& dest_fs, const RowsetId& new_rowset_id) = 0;
229
230
    virtual Status remove_old_files(std::vector<std::string>* files_to_remove) = 0;
231
232
    virtual Status check_file_exist() = 0;
233
234
40.0k
    bool need_delete_file() const { return _need_delete_file; }
235
236
40.3k
    void set_need_delete_file() { _need_delete_file = true; }
237
238
2.62M
    bool contains_version(Version version) const {
239
2.62M
        return rowset_meta()->version().contains(version);
240
2.62M
    }
241
242
1.95M
    static bool comparator(const RowsetSharedPtr& left, const RowsetSharedPtr& right) {
243
1.95M
        return left->end_version() < right->end_version();
244
1.95M
    }
245
246
    // this function is called by reader to increase reference of rowset
247
9.83M
    void acquire() { ++_refs_by_reader; }
248
249
9.82M
    void release() {
250
        // if the refs by reader is 0 and the rowset is closed, should release the resouce
251
9.82M
        uint64_t current_refs = --_refs_by_reader;
252
9.82M
        if (current_refs == 0 && _rowset_state_machine.rowset_state() == ROWSET_UNLOADING) {
253
0
            {
254
0
                std::lock_guard release_lock(_lock);
255
                // rejudge _refs_by_reader because we do not add lock in create reader
256
0
                if (_refs_by_reader == 0 &&
257
0
                    _rowset_state_machine.rowset_state() == ROWSET_UNLOADING) {
258
                    // first do close, then change state
259
0
                    do_close();
260
0
                    static_cast<void>(_rowset_state_machine.on_release());
261
0
                }
262
0
            }
263
0
            if (_rowset_state_machine.rowset_state() == ROWSET_UNLOADED) {
264
0
                VLOG_NOTICE
265
0
                        << "close the rowset. rowset state from ROWSET_UNLOADING to ROWSET_UNLOADED"
266
0
                        << ", version:" << start_version() << "-" << end_version()
267
0
                        << ", tabletid:" << _rowset_meta->tablet_id();
268
0
            }
269
0
        }
270
9.82M
    }
271
272
0
    void update_delayed_expired_timestamp(uint64_t delayed_expired_timestamp) {
273
0
        if (delayed_expired_timestamp > _delayed_expired_timestamp) {
274
0
            _delayed_expired_timestamp = delayed_expired_timestamp;
275
0
        }
276
0
    }
277
278
0
    uint64_t delayed_expired_timestamp() { return _delayed_expired_timestamp; }
279
280
3.68k
    virtual Status get_segments_key_bounds(std::vector<KeyBoundsPB>* segments_key_bounds) {
281
3.68k
        _rowset_meta->get_segments_key_bounds(segments_key_bounds);
282
3.68k
        return Status::OK();
283
3.68k
    }
284
285
3.66k
    void get_num_segment_rows(std::vector<uint32_t>* num_segment_rows) {
286
3.66k
        _rowset_meta->get_num_segment_rows(num_segment_rows);
287
3.66k
    }
288
289
    // min key of the first segment
290
1.59M
    bool first_key(std::string* min_key) {
291
1.59M
        KeyBoundsPB key_bounds;
292
1.59M
        bool ret = _rowset_meta->get_first_segment_key_bound(&key_bounds);
293
1.59M
        if (!ret) {
294
0
            return false;
295
0
        }
296
1.59M
        *min_key = key_bounds.min_key();
297
1.59M
        return true;
298
1.59M
    }
299
300
    // max key of the last segment
301
1.07M
    bool last_key(std::string* max_key) {
302
1.07M
        KeyBoundsPB key_bounds;
303
1.07M
        bool ret = _rowset_meta->get_last_segment_key_bound(&key_bounds);
304
1.07M
        if (!ret) {
305
0
            return false;
306
0
        }
307
1.07M
        *max_key = key_bounds.max_key();
308
1.07M
        return true;
309
1.07M
    }
310
311
1.59M
    bool is_segments_key_bounds_truncated() const {
312
1.59M
        return _rowset_meta->is_segments_key_bounds_truncated();
313
1.59M
    }
314
315
0
    bool is_segments_key_bounds_aggregated() const {
316
0
        return _rowset_meta->is_segments_key_bounds_aggregated();
317
0
    }
318
319
    bool check_rowset_segment();
320
321
0
    [[nodiscard]] virtual Status add_to_binlog() { return Status::OK(); }
322
323
    // is skip index compaction this time
324
11.2k
    bool is_skip_index_compaction(int32_t column_id) const {
325
11.2k
        return skip_index_compaction.find(column_id) != skip_index_compaction.end();
326
11.2k
    }
327
328
    // set skip index compaction next time
329
5
    void set_skip_index_compaction(int32_t column_id) { skip_index_compaction.insert(column_id); }
330
331
    std::string get_rowset_info_str();
332
333
    void clear_cache();
334
335
    MOCK_FUNCTION Result<std::string> segment_path(int64_t seg_id);
336
    RowsetSegmentView segment(size_t pos);
337
    RowsetSegmentRange segments();
338
339
    std::vector<std::string> get_index_file_names();
340
341
    // check if the rowset is a hole rowset
342
56.3k
    bool is_hole_rowset() const { return _is_hole_rowset; }
343
    // set the rowset as a hole rowset
344
1.24k
    void set_hole_rowset(bool is_hole_rowset) { _is_hole_rowset = is_hole_rowset; }
345
346
    int64_t approximate_cached_data_size();
347
348
    int64_t approximate_cache_index_size();
349
350
    std::chrono::time_point<std::chrono::system_clock> visible_timestamp() const;
351
352
protected:
353
    friend class RowsetFactory;
354
355
    DISALLOW_COPY_AND_ASSIGN(Rowset);
356
    // this is non-public because all clients should use RowsetFactory to obtain pointer to initialized Rowset
357
    Rowset(const TabletSchemaSPtr& schema, RowsetMetaSharedPtr rowset_meta,
358
           std::string tablet_path);
359
360
    // this is non-public because all clients should use RowsetFactory to obtain pointer to initialized Rowset
361
    virtual Status init() = 0;
362
363
    // release resources in this api
364
    virtual void do_close() = 0;
365
366
    virtual Status check_current_rowset_segment() = 0;
367
368
    virtual void clear_inverted_index_cache() = 0;
369
370
    TabletSchemaSPtr _schema;
371
372
    RowsetMetaSharedPtr _rowset_meta;
373
374
    // Local rowset requires a tablet path to obtain the absolute path on the local fs
375
    std::string _tablet_path;
376
377
    // init in constructor
378
    bool _is_pending;    // rowset is pending iff it's not in visible state
379
    bool _is_cumulative; // rowset is cumulative iff it's visible and start version < end version
380
381
    // mutex lock for load/close api because it is costly
382
    std::mutex _lock;
383
    bool _need_delete_file = false;
384
    // variable to indicate how many rowset readers owned this rowset
385
    std::atomic<uint64_t> _refs_by_reader;
386
    // rowset state machine
387
    RowsetStateMachine _rowset_state_machine;
388
    std::atomic<uint64_t> _delayed_expired_timestamp = 0;
389
390
    // <column_uniq_id>, skip index compaction
391
    std::set<int32_t> skip_index_compaction;
392
393
    // only used for cloud mode, it indicates whether this rowset is a hole rowset.
394
    // a hole rowset is a rowset that has no data, but is used to fill the version gap
395
    // it is used to ensure that the version sequence is continuous.
396
    bool _is_hole_rowset = false;
397
};
398
399
// Non-owning view over a segment in a Rowset. The referenced Rowset and RowsetMeta must outlive
400
// this view. Do not store it or capture it into async callbacks; copy the needed values or keep a
401
// RowsetSharedPtr instead.
402
class RowsetSegmentView {
403
public:
404
    using DeleteBitmapKey = std::tuple<RowsetId, uint32_t, int64_t>;
405
406
    RowsetSegmentView(Rowset* rowset, size_t pos)
407
5.26M
            : _rowset(rowset), _meta(rowset->rowset_meta()->segment(pos)) {}
408
409
    size_t pos() const { return _meta.pos(); }
410
2.79M
    int64_t id() const { return _meta.id(); }
411
2.43M
    RowsetSegmentRef ref() const { return _meta.ref(); }
412
0
    RowsetSegmentMetaView meta() const { return _meta; }
413
206k
    int64_t file_size() const { return _meta.file_size(); }
414
10.7k
    InvertedIndexFileInfo inverted_index_file_info() const {
415
10.7k
        return _meta.inverted_index_file_info();
416
10.7k
    }
417
0
    bool has_num_rows() const { return _meta.has_num_rows(); }
418
0
    int64_t num_rows() const { return _meta.num_rows(); }
419
0
    bool has_position_key_bounds() const { return _meta.has_position_key_bounds(); }
420
0
    const KeyBoundsPB& key_bounds() const { return _meta.key_bounds(); }
421
422
    std::string file_name() const;
423
    Result<std::string> path() const;
424
    io::UInt128Wrapper file_cache_key() const;
425
    DeleteBitmapKey delete_bitmap_key(int64_t version) const;
426
    RowLocation row_location(uint32_t row_id) const;
427
    std::vector<std::string> index_file_names() const;
428
    Result<std::string> index_file_cache_key(const TabletIndex& index) const;
429
430
private:
431
    Rowset* _rowset;
432
    RowsetSegmentMetaView _meta;
433
};
434
435
class RowsetSegmentRange {
436
public:
437
    class Iterator {
438
    public:
439
        using iterator_category = std::forward_iterator_tag;
440
        using value_type = RowsetSegmentView;
441
        using difference_type = std::ptrdiff_t;
442
443
8.58M
        Iterator(Rowset* rowset, size_t pos) : _rowset(rowset), _pos(pos) {}
444
445
2.63M
        RowsetSegmentView operator*() const { return {_rowset, _pos}; }
446
2.63M
        Iterator& operator++() {
447
2.63M
            ++_pos;
448
2.63M
            return *this;
449
2.63M
        }
450
6.92M
        bool operator==(const Iterator& other) const {
451
6.92M
            return _rowset == other._rowset && _pos == other._pos;
452
6.92M
        }
453
6.92M
        bool operator!=(const Iterator& other) const { return !(*this == other); }
454
455
    private:
456
        Rowset* _rowset;
457
        size_t _pos;
458
    };
459
460
4.28M
    explicit RowsetSegmentRange(Rowset* rowset) : _rowset(rowset) {}
461
462
4.29M
    Iterator begin() const { return {_rowset, 0}; }
463
4.29M
    Iterator end() const { return {_rowset, cast_set<size_t>(_rowset->num_segments())}; }
464
465
private:
466
    Rowset* _rowset;
467
};
468
469
2.63M
inline RowsetSegmentView Rowset::segment(size_t pos) {
470
2.63M
    return RowsetSegmentView(this, pos);
471
2.63M
}
472
473
4.28M
inline RowsetSegmentRange Rowset::segments() {
474
4.28M
    return RowsetSegmentRange(this);
475
4.28M
}
476
477
// `rs_metas` MUST already be sorted by `RowsetMeta::comparator`
478
Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);
479
480
} // namespace doris