Coverage Report

Created: 2026-08-27 08:50

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
985k
    RowsetStateMachine() : _rowset_state(ROWSET_UNLOADED) {}
77
78
285k
    Status on_load() {
79
285k
        switch (_rowset_state) {
80
285k
        case ROWSET_UNLOADED:
81
285k
            _rowset_state = ROWSET_LOADED;
82
285k
            break;
83
84
0
        default:
85
0
            return Status::Error<ErrorCode::ROWSET_INVALID_STATE_TRANSITION>(
86
0
                    "RowsetStateMachine meet invalid state");
87
285k
        }
88
285k
        return Status::OK();
89
285k
    }
90
91
4
    Status on_close(uint64_t refs_by_reader) {
92
4
        switch (_rowset_state) {
93
4
        case ROWSET_LOADED:
94
4
            if (refs_by_reader == 0) {
95
4
                _rowset_state = ROWSET_UNLOADED;
96
4
            } else {
97
0
                _rowset_state = ROWSET_UNLOADING;
98
0
            }
99
4
            break;
100
101
0
        default:
102
0
            return Status::Error<ErrorCode::ROWSET_INVALID_STATE_TRANSITION>(
103
0
                    "RowsetStateMachine meet invalid state");
104
4
        }
105
4
        return Status::OK();
106
4
    }
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
8.60M
    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
124M
    const RowsetMetaSharedPtr& rowset_meta() const { return _rowset_meta; }
140
141
    void merge_rowset_meta(const RowsetMeta& other);
142
143
6
    bool is_pending() const { return _is_pending; }
144
145
17.3M
    bool is_local() const { return _rowset_meta->is_local(); }
146
147
    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
6.48M
    int64_t start_version() const { return rowset_meta()->version().first; }
156
12.9M
    int64_t end_version() const { return rowset_meta()->version().second; }
157
236k
    int64_t index_disk_size() const { return rowset_meta()->index_disk_size(); }
158
7.01M
    int64_t data_disk_size() const { return rowset_meta()->data_disk_size(); }
159
555k
    int64_t total_disk_size() const { return rowset_meta()->total_disk_size(); }
160
8.45k
    bool empty() const { return rowset_meta()->empty(); }
161
0
    bool zero_num_rows() const { return rowset_meta()->num_rows() == 0; }
162
6.19M
    size_t num_rows() const { return rowset_meta()->num_rows(); }
163
20.6M
    Version version() const { return rowset_meta()->version(); }
164
22.5M
    RowsetId rowset_id() const { return rowset_meta()->rowset_id(); }
165
128k
    int64_t creation_time() const { return rowset_meta()->creation_time(); }
166
0
    PUniqueId load_id() const { return rowset_meta()->load_id(); }
167
152k
    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
14.3M
    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
384k
    int64_t newest_write_timestamp() const { return rowset_meta()->newest_write_timestamp(); }
176
    // The commit tso range of the data in rowset.
177
208k
    TsoRange commit_tso() const { return rowset_meta()->commit_tso(); }
178
179
1.35M
    bool is_segments_overlapping() const { return rowset_meta()->is_segments_overlapping(); }
180
2.91M
    KeysType keys_type() { return _schema->keys_type(); }
181
3.25k
    RowsetStatePB rowset_meta_state() const { return rowset_meta()->rowset_state(); }
182
41
    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
12.6k
    void close() {
192
12.6k
        RowsetState old_state = _rowset_state_machine.rowset_state();
193
12.6k
        if (old_state != ROWSET_LOADED) {
194
12.6k
            return;
195
12.6k
        }
196
4
        Status st = Status::OK();
197
4
        {
198
4
            std::lock_guard close_lock(_lock);
199
4
            uint64_t current_refs = _refs_by_reader;
200
4
            old_state = _rowset_state_machine.rowset_state();
201
4
            if (old_state != ROWSET_LOADED) {
202
0
                return;
203
0
            }
204
4
            if (current_refs == 0) {
205
4
                do_close();
206
4
            }
207
4
            st = _rowset_state_machine.on_close(current_refs);
208
4
        }
209
4
        if (!st.ok()) {
210
0
            LOG(WARNING) << "state transition failed from:" << _rowset_state_machine.rowset_state();
211
0
            return;
212
0
        }
213
4
        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
4
    }
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
12.5k
    bool need_delete_file() const { return _need_delete_file; }
235
236
12.6k
    void set_need_delete_file() { _need_delete_file = true; }
237
238
2.26k
    bool contains_version(Version version) const {
239
2.26k
        return rowset_meta()->version().contains(version);
240
2.26k
    }
241
242
1.10M
    static bool comparator(const RowsetSharedPtr& left, const RowsetSharedPtr& right) {
243
1.10M
        return left->end_version() < right->end_version();
244
1.10M
    }
245
246
    // this function is called by reader to increase reference of rowset
247
6.66M
    void acquire() { ++_refs_by_reader; }
248
249
6.66M
    void release() {
250
        // if the refs by reader is 0 and the rowset is closed, should release the resouce
251
6.66M
        uint64_t current_refs = --_refs_by_reader;
252
6.66M
        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
6.66M
    }
271
272
81
    virtual Status get_segments_key_bounds(std::vector<KeyBoundsPB>* segments_key_bounds) {
273
81
        _rowset_meta->get_segments_key_bounds(segments_key_bounds);
274
81
        return Status::OK();
275
81
    }
276
277
54
    void get_num_segment_rows(std::vector<uint32_t>* num_segment_rows) {
278
54
        _rowset_meta->get_num_segment_rows(num_segment_rows);
279
54
    }
280
281
    // min key of the first segment
282
984k
    bool first_key(std::string* min_key) {
283
984k
        KeyBoundsPB key_bounds;
284
984k
        bool ret = _rowset_meta->get_first_segment_key_bound(&key_bounds);
285
984k
        if (!ret) {
286
0
            return false;
287
0
        }
288
984k
        *min_key = key_bounds.min_key();
289
984k
        return true;
290
984k
    }
291
292
    // max key of the last segment
293
650k
    bool last_key(std::string* max_key) {
294
650k
        KeyBoundsPB key_bounds;
295
650k
        bool ret = _rowset_meta->get_last_segment_key_bound(&key_bounds);
296
650k
        if (!ret) {
297
0
            return false;
298
0
        }
299
650k
        *max_key = key_bounds.max_key();
300
650k
        return true;
301
650k
    }
302
303
985k
    bool is_segments_key_bounds_truncated() const {
304
985k
        return _rowset_meta->is_segments_key_bounds_truncated();
305
985k
    }
306
307
0
    bool is_segments_key_bounds_aggregated() const {
308
0
        return _rowset_meta->is_segments_key_bounds_aggregated();
309
0
    }
310
311
    bool check_rowset_segment();
312
313
0
    [[nodiscard]] virtual Status add_to_binlog() { return Status::OK(); }
314
315
    // is skip index compaction this time
316
12.7k
    bool is_skip_index_compaction(int32_t column_id) const {
317
12.7k
        return skip_index_compaction.find(column_id) != skip_index_compaction.end();
318
12.7k
    }
319
320
    // set skip index compaction next time
321
9
    void set_skip_index_compaction(int32_t column_id) { skip_index_compaction.insert(column_id); }
322
323
    std::string get_rowset_info_str();
324
325
    void clear_cache();
326
327
    MOCK_FUNCTION Result<std::string> segment_path(int64_t seg_id);
328
    RowsetSegmentView segment(size_t pos);
329
    RowsetSegmentRange segments();
330
331
    std::vector<std::string> get_index_file_names();
332
333
    // check if the rowset is a hole rowset
334
71.8k
    bool is_hole_rowset() const { return _is_hole_rowset; }
335
    // set the rowset as a hole rowset
336
1.28k
    void set_hole_rowset(bool is_hole_rowset) { _is_hole_rowset = is_hole_rowset; }
337
338
    int64_t approximate_cached_data_size();
339
340
    int64_t approximate_cache_index_size();
341
342
    std::chrono::time_point<std::chrono::system_clock> visible_timestamp() const;
343
344
protected:
345
    friend class RowsetFactory;
346
347
    DISALLOW_COPY_AND_ASSIGN(Rowset);
348
    // this is non-public because all clients should use RowsetFactory to obtain pointer to initialized Rowset
349
    Rowset(const TabletSchemaSPtr& schema, RowsetMetaSharedPtr rowset_meta,
350
           std::string tablet_path);
351
352
    // this is non-public because all clients should use RowsetFactory to obtain pointer to initialized Rowset
353
    virtual Status init() = 0;
354
355
    // release resources in this api
356
    virtual void do_close() = 0;
357
358
    virtual Status check_current_rowset_segment() = 0;
359
360
    virtual void clear_inverted_index_cache() = 0;
361
362
    TabletSchemaSPtr _schema;
363
364
    RowsetMetaSharedPtr _rowset_meta;
365
366
    // Local rowset requires a tablet path to obtain the absolute path on the local fs
367
    std::string _tablet_path;
368
369
    // init in constructor
370
    bool _is_pending;    // rowset is pending iff it's not in visible state
371
    bool _is_cumulative; // rowset is cumulative iff it's visible and start version < end version
372
373
    // mutex lock for load/close api because it is costly
374
    std::mutex _lock;
375
    bool _need_delete_file = false;
376
    // variable to indicate how many rowset readers owned this rowset
377
    std::atomic<uint64_t> _refs_by_reader;
378
    // rowset state machine
379
    RowsetStateMachine _rowset_state_machine;
380
    // <column_uniq_id>, skip index compaction
381
    std::set<int32_t> skip_index_compaction;
382
383
    // only used for cloud mode, it indicates whether this rowset is a hole rowset.
384
    // a hole rowset is a rowset that has no data, but is used to fill the version gap
385
    // it is used to ensure that the version sequence is continuous.
386
    bool _is_hole_rowset = false;
387
};
388
389
// Non-owning view over a segment in a Rowset. The referenced Rowset and RowsetMeta must outlive
390
// this view. Do not store it or capture it into async callbacks; copy the needed values or keep a
391
// RowsetSharedPtr instead.
392
class RowsetSegmentView {
393
public:
394
    using DeleteBitmapKey = std::tuple<RowsetId, uint32_t, int64_t>;
395
396
    RowsetSegmentView(Rowset* rowset, size_t pos)
397
3.56M
            : _rowset(rowset), _meta(rowset->rowset_meta()->segment(pos)) {}
398
399
    size_t pos() const { return _meta.pos(); }
400
1.83M
    int64_t id() const { return _meta.id(); }
401
1.68M
    RowsetSegmentRef ref() const { return _meta.ref(); }
402
0
    RowsetSegmentMetaView meta() const { return _meta; }
403
203k
    int64_t file_size() const { return _meta.file_size(); }
404
10.7k
    InvertedIndexFileInfo inverted_index_file_info() const {
405
10.7k
        return _meta.inverted_index_file_info();
406
10.7k
    }
407
0
    bool has_num_rows() const { return _meta.has_num_rows(); }
408
0
    int64_t num_rows() const { return _meta.num_rows(); }
409
0
    bool has_position_key_bounds() const { return _meta.has_position_key_bounds(); }
410
0
    const KeyBoundsPB& key_bounds() const { return _meta.key_bounds(); }
411
412
    std::string file_name() const;
413
    Result<std::string> path() const;
414
    io::UInt128Wrapper file_cache_key() const;
415
    DeleteBitmapKey delete_bitmap_key(int64_t version) const;
416
    RowLocation row_location(uint32_t row_id) const;
417
    std::vector<std::string> index_file_names() const;
418
    Result<std::string> index_file_cache_key(const TabletIndex& index) const;
419
420
private:
421
    Rowset* _rowset;
422
    RowsetSegmentMetaView _meta;
423
};
424
425
class RowsetSegmentRange {
426
public:
427
    class Iterator {
428
    public:
429
        using iterator_category = std::forward_iterator_tag;
430
        using value_type = RowsetSegmentView;
431
        using difference_type = std::ptrdiff_t;
432
433
6.25M
        Iterator(Rowset* rowset, size_t pos) : _rowset(rowset), _pos(pos) {}
434
435
1.73M
        RowsetSegmentView operator*() const { return {_rowset, _pos}; }
436
1.73M
        Iterator& operator++() {
437
1.73M
            ++_pos;
438
1.73M
            return *this;
439
1.73M
        }
440
4.86M
        bool operator==(const Iterator& other) const {
441
4.86M
            return _rowset == other._rowset && _pos == other._pos;
442
4.86M
        }
443
4.86M
        bool operator!=(const Iterator& other) const { return !(*this == other); }
444
445
    private:
446
        Rowset* _rowset;
447
        size_t _pos;
448
    };
449
450
3.12M
    explicit RowsetSegmentRange(Rowset* rowset) : _rowset(rowset) {}
451
452
3.13M
    Iterator begin() const { return {_rowset, 0}; }
453
3.13M
    Iterator end() const { return {_rowset, cast_set<size_t>(_rowset->num_segments())}; }
454
455
private:
456
    Rowset* _rowset;
457
};
458
459
1.82M
inline RowsetSegmentView Rowset::segment(size_t pos) {
460
1.82M
    return RowsetSegmentView(this, pos);
461
1.82M
}
462
463
3.12M
inline RowsetSegmentRange Rowset::segments() {
464
3.12M
    return RowsetSegmentRange(this);
465
3.12M
}
466
467
// `rs_metas` MUST already be sorted by `RowsetMeta::comparator`
468
Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);
469
470
} // namespace doris