Coverage Report

Created: 2026-10-09 12:56

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/load/memtable/memtable.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 <stddef.h>
21
#include <stdint.h>
22
23
#include <cstdint>
24
#include <cstring>
25
#include <functional>
26
#include <memory>
27
#include <vector>
28
29
#include "common/status.h"
30
#include "core/arena.h"
31
#include "core/block/block.h"
32
#include "core/custom_allocator.h"
33
#include "exprs/aggregate/aggregate_function.h"
34
#include "runtime/memory/mem_tracker.h"
35
#include "runtime/thread_context.h"
36
#include "storage/partial_update_info.h"
37
#include "storage/rowset/rowset_fwd.h"
38
#include "storage/tablet/tablet_schema.h"
39
40
namespace doris {
41
42
class SlotDescriptor;
43
class TabletSchema;
44
struct TabletAddRowsPayload;
45
class TupleDescriptor;
46
enum KeysType : int;
47
48
// Active: the memtable is currently used by writer to insert into blocks
49
// Write_finished: the memtable finished write blocks and in the queue waiting for flush
50
// FLUSH: the memtable is under flushing, write segment to disk.
51
enum MemType { ACTIVE = 0, WRITE_FINISHED = 1, FLUSH = 2 };
52
53
// row pos in _input_mutable_block
54
struct RowInBlock {
55
    size_t _row_pos;
56
    int64_t _allocated_lsn = 0;
57
    char* _agg_mem = nullptr;
58
    size_t* _agg_state_offset = nullptr;
59
    bool _has_init_agg;
60
61
34.8k
    RowInBlock(size_t row = 0) : _row_pos(row), _has_init_agg(false) {}
62
    RowInBlock(size_t row, int64_t allocated_lsn)
63
26.6k
            : _row_pos(row), _allocated_lsn(allocated_lsn), _has_init_agg(false) {}
64
65
2.05k
    void init_agg_places(char* agg_mem, size_t* agg_state_offset) {
66
2.05k
        _has_init_agg = true;
67
2.05k
        _agg_mem = agg_mem;
68
2.05k
        _agg_state_offset = agg_state_offset;
69
2.05k
    }
70
71
30.7k
    char* agg_places(size_t offset) const { return _agg_mem + _agg_state_offset[offset]; }
72
73
38.9k
    inline bool has_init_agg() const { return _has_init_agg; }
74
75
2.05k
    inline void remove_init_agg() { _has_init_agg = false; }
76
};
77
78
class Tie {
79
public:
80
    class Iter {
81
    public:
82
135
        Iter(Tie& tie) : _tie(tie), _next(tie._begin + 1) {}
83
10.3k
        size_t left() const { return _left; }
84
26.7k
        size_t right() const { return _right; }
85
86
        // return false means no more ranges
87
4.28k
        bool next() {
88
4.28k
            if (_next >= _tie._end) {
89
92
                return false;
90
92
            }
91
4.19k
            _next = _find(1, _next);
92
4.19k
            if (_next >= _tie._end) {
93
44
                return false;
94
44
            }
95
4.14k
            _left = _next - 1;
96
4.14k
            _next = _find(0, _next);
97
4.14k
            _right = _next;
98
4.14k
            return true;
99
4.19k
        }
100
101
    private:
102
8.34k
        size_t _find(uint8_t value, size_t start) {
103
8.34k
            if (start >= _tie._end) {
104
0
                return start;
105
0
            }
106
8.34k
            size_t offset = start - _tie._begin;
107
8.34k
            size_t size = _tie._end - start;
108
8.34k
            void* p = std::memchr(_tie._bits.data() + offset, value, size);
109
8.34k
            if (p == nullptr) {
110
95
                return _tie._end;
111
95
            }
112
8.24k
            return static_cast<uint8_t*>(p) - _tie._bits.data() + _tie._begin;
113
8.34k
        }
114
115
    private:
116
        Tie& _tie;
117
        size_t _left;
118
        size_t _right;
119
        size_t _next;
120
    };
121
122
public:
123
49
    Tie(size_t begin, size_t end) : _begin(begin), _end(end), _bits(end - begin, 1) {}
124
0
    uint8_t operator[](size_t i) const { return _bits[i - _begin]; }
125
20.5k
    uint8_t& operator[](size_t i) { return _bits[i - _begin]; }
126
135
    Iter iter() { return Iter(*this); }
127
128
private:
129
    const size_t _begin;
130
    const size_t _end;
131
    DorisVector<uint8_t> _bits;
132
};
133
134
class RowInBlockComparator {
135
public:
136
    RowInBlockComparator(std::shared_ptr<TabletSchema> tablet_schema)
137
50
            : _tablet_schema(tablet_schema) {}
138
    // call set_block before operator().
139
    // only first time insert block to create _input_mutable_block,
140
    // so can not Comparator of construct to set pblock
141
89
    void set_block(MutableBlock* pblock) { _pblock = pblock; }
142
    int operator()(const RowInBlock* left, const RowInBlock* right) const;
143
144
private:
145
    std::shared_ptr<TabletSchema> _tablet_schema;
146
    MutableBlock* _pblock = nullptr; //  corresponds to Memtable::_input_mutable_block
147
};
148
149
class MemTableStat {
150
public:
151
14
    MemTableStat& operator+=(const MemTableStat& stat) {
152
14
        raw_rows += stat.raw_rows;
153
14
        merged_rows += stat.merged_rows;
154
14
        sort_ns += stat.sort_ns;
155
14
        agg_ns += stat.agg_ns;
156
14
        put_into_output_ns += stat.put_into_output_ns;
157
14
        duration_ns += stat.duration_ns;
158
14
        sort_times += stat.sort_times;
159
14
        agg_times += stat.agg_times;
160
161
14
        return *this;
162
14
    }
163
164
    std::atomic<int64_t> raw_rows = 0;
165
    std::atomic<int64_t> merged_rows = 0;
166
    int64_t sort_ns = 0;
167
    int64_t agg_ns = 0;
168
    int64_t put_into_output_ns = 0;
169
    int64_t duration_ns = 0;
170
    std::atomic<int64_t> sort_times = 0;
171
    std::atomic<int64_t> agg_times = 0;
172
};
173
174
class MemTable {
175
public:
176
    MemTable(int64_t tablet_id, std::shared_ptr<TabletSchema> tablet_schema,
177
             const std::vector<SlotDescriptor*>* slot_descs, TupleDescriptor* tuple_desc,
178
             bool enable_unique_key_mow, PartialUpdateInfo* partial_update_info,
179
             const std::shared_ptr<ResourceContext>& resource_ctx, bool need_lsn = false);
180
    ~MemTable();
181
182
23
    int64_t tablet_id() const { return _tablet_id; }
183
114
    size_t memory_usage() const { return _mem_tracker->consumption(); }
184
    size_t get_flush_reserve_memory_size() const;
185
    // insert tuple from (row_pos) to (row_pos+num_rows)
186
    Status insert(const Block* block, const TabletAddRowsPayload& rows);
187
188
    void shrink_memtable_by_agg();
189
190
    bool need_flush() const;
191
192
    bool need_agg() const;
193
194
    Status to_block(std::unique_ptr<Block>* res);
195
196
5
    ConstAllocatedLsnVectorSharedPtr allocated_lsns() const { return _output_allocated_lsns; }
197
198
33
    bool empty() const { return _input_mutable_block.rows() == 0; }
199
200
14
    const MemTableStat& stat() { return _stat; }
201
202
75
    std::shared_ptr<ResourceContext> resource_ctx() { return _resource_ctx; }
203
204
58
    std::shared_ptr<MemTracker> mem_tracker() { return _mem_tracker; }
205
206
19
    void set_flush_success() { _is_flush_success = true; }
207
208
0
    MemType get_mem_type() { return _mem_type; }
209
210
42
    void update_mem_type(MemType memtype) { _mem_type = memtype; }
211
212
34
    int64_t raw_rows() { return _stat.raw_rows.load(); }
213
214
private:
215
    // for vectorized
216
    template <bool has_skip_bitmap_col>
217
    void _aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlock* new_row,
218
                                     RowInBlock* row_in_skiplist);
219
220
    // Merge allocated LSN sidecar only when MemTable merges two RowInBlock objects.
221
    // Table models that require complex merge semantics, such as AGG tables and unique key
222
    // merge-on-read tables, do not support allocated LSN now and are rejected in insert().
223
    void _merge_allocated_lsn(RowInBlock* src_row, RowInBlock* dst_row);
224
225
    void _append_output_allocated_lsn(RowInBlock* row);
226
227
    void _aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, RowInBlock* new_row,
228
                                              RowInBlock* row_in_skiplist);
229
230
    // Used to wrapped by to_block to do exception handle logic
231
    Status _to_block(std::unique_ptr<Block>* res);
232
233
    int64_t _adaptive_write_buffer_size() const;
234
235
private:
236
    std::atomic<MemType> _mem_type;
237
    int64_t _tablet_id;
238
    bool _enable_unique_key_mow = false;
239
    bool _is_flush_success = false;
240
    UniqueKeyUpdateModePB _partial_update_mode {UniqueKeyUpdateModePB::UPSERT};
241
    const KeysType _keys_type;
242
    std::shared_ptr<TabletSchema> _tablet_schema;
243
244
    std::shared_ptr<RowInBlockComparator> _vec_row_comparator;
245
246
    std::shared_ptr<ResourceContext> _resource_ctx;
247
248
    std::shared_ptr<MemTracker> _mem_tracker;
249
    // Only the rows will be inserted into block can allocate memory from _arena.
250
    // In this way, we can make MemTable::memory_usage() to be more accurate, and eventually
251
    // reduce the number of segment files that are generated by current load
252
    Arena _arena;
253
    int64_t _load_mem_limit = -1;
254
255
    void _init_columns_offset_by_slot_descs(const std::vector<SlotDescriptor*>* slot_descs,
256
                                            const TupleDescriptor* tuple_desc);
257
    std::vector<int> _column_offset;
258
    int32_t _row_lsn_col_pos = -1;
259
260
    // Number of rows inserted to this memtable.
261
    // This is not the rows in this memtable, because rows may be merged
262
    // in unique or aggregate key model.
263
    MemTableStat _stat;
264
265
    //for vectorized
266
    MutableBlock _input_mutable_block;
267
    MutableBlock _output_mutable_block;
268
    AllocatedLsnVectorSharedPtr _output_allocated_lsns = std::make_shared<std::vector<int64_t>>();
269
    bool _need_lsn = false;
270
    size_t _last_sorted_pos = 0;
271
    size_t _last_agg_pos = 0;
272
273
    //return number of same keys
274
    size_t _sort();
275
    Status _sort_by_cluster_keys();
276
    void _sort_one_column(DorisVector<std::shared_ptr<RowInBlock>>& row_in_blocks, Tie& tie,
277
                          std::function<int(RowInBlock*, RowInBlock*)> cmp);
278
    template <bool is_final>
279
    void _finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, int row_pos);
280
    void _init_row_for_agg(RowInBlock* row, MutableBlock& mutable_block);
281
    void _clear_row_agg(RowInBlock* row);
282
283
    template <bool is_final, bool has_skip_bitmap_col = false>
284
    void _aggregate();
285
286
    template <bool is_final>
287
    void _aggregate_for_flexible_partial_update_without_seq_col(
288
            MutableBlock& mutable_block,
289
            DorisVector<std::shared_ptr<RowInBlock>>& temp_row_in_blocks);
290
291
    template <bool is_final>
292
    void _aggregate_for_flexible_partial_update_with_seq_col(
293
            MutableBlock& mutable_block,
294
            DorisVector<std::shared_ptr<RowInBlock>>& temp_row_in_blocks);
295
296
    Status _put_into_output(Block& in_block);
297
    bool _is_first_insertion;
298
299
    void _init_agg_functions(const Block* block);
300
    std::vector<AggregateFunctionPtr> _agg_functions;
301
    std::vector<size_t> _offsets_of_aggregate_states;
302
    size_t _total_size_of_aggregate_states;
303
    std::unique_ptr<DorisVector<std::shared_ptr<RowInBlock>>> _row_in_blocks;
304
305
    size_t _num_columns;
306
    int32_t _seq_col_idx_in_block {-1};
307
    int32_t _skip_bitmap_col_idx {-1};
308
    int32_t _delete_sign_col_idx {-1};
309
    int32_t _delete_sign_col_unique_id {-1};
310
    int32_t _seq_col_unique_id {-1};
311
312
    bool _is_partial_update_and_auto_inc = false;
313
}; // class MemTable
314
315
} // namespace doris