Coverage Report

Created: 2026-09-15 16:32

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/txn/txn_manager.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 <gen_cpp/Types_types.h>
22
#include <gen_cpp/types.pb.h>
23
#include <stddef.h>
24
#include <stdint.h>
25
26
#include <boost/container/detail/std_fwd.hpp>
27
#include <map>
28
#include <memory>
29
#include <mutex>
30
#include <set>
31
#include <shared_mutex>
32
#include <unordered_map>
33
#include <unordered_set>
34
#include <utility>
35
#include <vector>
36
37
#include "common/status.h"
38
#include "core/block/block.h"
39
#include "runtime/memory/lru_cache_policy.h"
40
#include "storage/olap_common.h"
41
#include "storage/rowset/pending_rowset_helper.h"
42
#include "storage/rowset/rowset.h"
43
#include "storage/rowset/rowset_meta.h"
44
#include "storage/segment/segment.h"
45
#include "storage/segment/vertical_segment_writer.h"
46
#include "storage/tablet/tablet.h"
47
#include "storage/tablet/tablet_meta.h"
48
#include "storage/tablet_info.h"
49
#include "util/time.h"
50
51
namespace doris {
52
class OlapMeta;
53
struct TabletPublishStatistics;
54
struct PartialUpdateInfo;
55
56
enum class TxnState {
57
    NOT_FOUND = 0,
58
    PREPARED = 1,
59
    COMMITTED = 2,
60
    ROLLEDBACK = 3,
61
    ABORTED = 4,
62
    DELETED = 5,
63
};
64
enum class PublishStatus { INIT = 0, PREPARE = 1, SUCCEED = 2 };
65
66
// The row binlog rowset and its independent binlog tablet, carried through commit and publish.
67
struct RowBinlogTxnInfo {
68
    RowsetSharedPtr rowset;
69
    BaseTabletSPtr tablet;
70
    // Delete bitmap deltas that should be applied to the independent binlog tablet.
71
    DeleteBitmapPtr delete_bitmap;
72
    // Immutable in-memory snapshot, independent of the writer and bitmap LRU. Null means absent.
73
    std::shared_ptr<const PRowBinlogWriteColumnMappings> column_mapping_snapshot;
74
};
75
76
struct TxnPublishInfo {
77
    int64_t publish_version {-1};
78
    int64_t base_compaction_cnt {-1};
79
    int64_t cumulative_compaction_cnt {-1};
80
    int64_t cumulative_point {-1};
81
};
82
83
struct TabletTxnInfo {
84
    PUniqueId load_id;
85
    RowsetSharedPtr rowset;
86
    // The row-binlog tablet is attached while PREPARED; its rowset is filled at commit.
87
    RowBinlogTxnInfo attach_row_binlog;
88
    PendingRowsetGuard pending_rs_guard;
89
    bool unique_key_merge_on_write {false};
90
    DeleteBitmapPtr delete_bitmap;
91
    // records rowsets calc in commit txn
92
    RowsetIdUnorderedSet rowset_ids;
93
    int64_t creation_time;
94
    bool ingest {false};
95
    std::shared_ptr<PartialUpdateInfo> partial_update_info;
96
97
    // for cloud only, used to determine if a retry CloudTabletCalcDeleteBitmapTask
98
    // needs to re-calculate the delete bitmap
99
    std::shared_ptr<PublishStatus> publish_status;
100
    TxnPublishInfo publish_info;
101
102
    // for cloud only, used to calculate delete bitmap for txn load
103
    bool is_txn_load = false;
104
    std::vector<RowsetSharedPtr> invisible_rowsets;
105
    int64_t lock_id;
106
    int64_t next_visible_version;
107
108
    TxnState state {TxnState::PREPARED};
109
23.1k
    TabletTxnInfo() = default;
110
111
    TabletTxnInfo(PUniqueId load_id, RowsetSharedPtr rowset)
112
472
            : load_id(std::move(load_id)),
113
472
              rowset(std::move(rowset)),
114
472
              creation_time(UnixSeconds()) {}
115
116
    TabletTxnInfo(PUniqueId load_id, RowsetSharedPtr rowset, bool ingest_arg)
117
317
            : load_id(std::move(load_id)),
118
317
              rowset(std::move(rowset)),
119
317
              creation_time(UnixSeconds()),
120
317
              ingest(ingest_arg) {}
121
122
    TabletTxnInfo(PUniqueId load_id, RowsetSharedPtr rowset, bool merge_on_write,
123
                  DeleteBitmapPtr delete_bitmap, RowsetIdUnorderedSet ids)
124
            : load_id(std::move(load_id)),
125
              rowset(std::move(rowset)),
126
              unique_key_merge_on_write(merge_on_write),
127
              delete_bitmap(std::move(delete_bitmap)),
128
              rowset_ids(std::move(ids)),
129
0
              creation_time(UnixSeconds()) {}
130
131
317
    void prepare() { state = TxnState::PREPARED; }
132
472
    void commit() { state = TxnState::COMMITTED; }
133
0
    void rollback() { state = TxnState::ROLLEDBACK; }
134
0
    void abort() {
135
0
        if (state == TxnState::PREPARED) {
136
0
            state = TxnState::ABORTED;
137
0
        }
138
0
    }
139
};
140
141
struct CommitTabletTxnInfo {
142
    TTransactionId transaction_id {0};
143
    TPartitionId partition_id {0};
144
    DeleteBitmapPtr delete_bitmap;
145
    RowsetIdUnorderedSet rowset_ids;
146
    std::shared_ptr<PartialUpdateInfo> partial_update_info;
147
};
148
149
using CommitTabletTxnInfoVec = std::vector<CommitTabletTxnInfo>;
150
151
// txn manager is used to manage mapping between tablet and txns
152
class TxnManager {
153
public:
154
    TxnManager(StorageEngine& engine, int32_t txn_map_shard_size, int32_t txn_shard_size);
155
156
700
    ~TxnManager() {
157
700
        delete[] _txn_tablet_maps;
158
700
        delete[] _txn_partition_maps;
159
700
        delete[] _txn_map_locks;
160
700
        delete[] _txn_mutex;
161
700
    }
162
163
    class CacheValue : public LRUCacheValueBase {
164
    public:
165
        int64_t value;
166
    };
167
168
    // add a txn to manager
169
    // partition id is useful in publish version stage because version is associated with partition
170
    Status prepare_txn(TPartitionId partition_id, const Tablet& tablet,
171
                       TTransactionId transaction_id, const PUniqueId& load_id,
172
                       bool is_ingest = false);
173
    // most used for ut
174
    Status prepare_txn(TPartitionId partition_id, TTransactionId transaction_id,
175
                       TTabletId tablet_id, TabletUid tablet_uid, const PUniqueId& load_id,
176
                       bool is_ingest = false);
177
178
    // Internal: attach the independent row-binlog tablet to an existing base-tablet transaction
179
    // before commit, so tablet lifecycle operations can drain the whole group transaction.
180
    Status attach_row_binlog_tablet_to_txn(TPartitionId partition_id, TTransactionId transaction_id,
181
                                           const TabletInfo& base_tablet_info,
182
                                           const BaseTabletSPtr& row_binlog_tablet);
183
184
    Status commit_txn(TPartitionId partition_id, const Tablet& tablet,
185
                      TTransactionId transaction_id, const PUniqueId& load_id,
186
                      const RowsetSharedPtr& rowset_ptr, PendingRowsetGuard guard, bool is_recovery,
187
                      std::shared_ptr<PartialUpdateInfo> partial_update_info = nullptr,
188
                      const RowBinlogTxnInfo& attach_row_binlog = {});
189
190
    Status publish_txn(
191
            TPartitionId partition_id, const TabletSharedPtr& tablet, TTransactionId transaction_id,
192
            const Version& version, TabletPublishStatistics* stats,
193
            std::shared_ptr<TabletTxnInfo>& extend_tablet_txn_info, const int64_t commit_tso = -1,
194
            std::shared_ptr<const PRowBinlogWriteColumnMappings> row_binlog_column_mappings =
195
                    nullptr);
196
197
    // delete the txn from manager if it is not committed(not have a valid rowset)
198
    Status rollback_txn(TPartitionId partition_id, const Tablet& tablet,
199
                        TTransactionId transaction_id);
200
201
    Status delete_txn(TPartitionId partition_id, const TabletSharedPtr& tablet,
202
                      TTransactionId transaction_id);
203
204
    Status commit_txn(OlapMeta* meta, TPartitionId partition_id, TTransactionId transaction_id,
205
                      TTabletId tablet_id, TabletUid tablet_uid, const PUniqueId& load_id,
206
                      const RowsetSharedPtr& rowset_ptr, PendingRowsetGuard guard, bool is_recovery,
207
                      std::shared_ptr<PartialUpdateInfo> partial_update_info = nullptr,
208
                      const RowBinlogTxnInfo& attach_row_binlog = {});
209
210
    // remove a txn from txn manager
211
    // not persist rowset meta because
212
    Status publish_txn(OlapMeta* meta, TPartitionId partition_id, TTransactionId transaction_id,
213
                       TTabletId tablet_id, TabletUid tablet_uid, const Version& version,
214
                       TabletPublishStatistics* stats,
215
                       std::shared_ptr<TabletTxnInfo>& extend_tablet_txn_info,
216
                       const int64_t commit_tso = -1,
217
                       std::shared_ptr<const PRowBinlogWriteColumnMappings>
218
                               row_binlog_column_mappings = nullptr);
219
220
    // only abort not committed txn
221
    void abort_txn(TPartitionId partition_id, TTransactionId transaction_id, TTabletId tablet_id,
222
                   TabletUid tablet_uid);
223
224
    // delete the txn from manager if it is not committed(not have a valid rowset)
225
    Status rollback_txn(TPartitionId partition_id, TTransactionId transaction_id,
226
                        TTabletId tablet_id, TabletUid tablet_uid);
227
228
    // remove the txn from txn manager
229
    // delete the related rowset if it is not null
230
    // delete rowset related data if it is not null
231
    Status delete_txn(OlapMeta* meta, TPartitionId partition_id, TTransactionId transaction_id,
232
                      TTabletId tablet_id, TabletUid tablet_uid);
233
234
    // Includes transactions keyed by this tablet and base-tablet transactions that attach this
235
    // tablet as their independent row-binlog tablet.
236
    void get_tablet_related_txns(TTabletId tablet_id, TabletUid tablet_uid, int64_t* partition_id,
237
                                 std::set<int64_t>* transaction_ids);
238
239
    void get_txn_related_tablets(
240
            const TTransactionId transaction_id, TPartitionId partition_ids,
241
            std::map<TabletInfo, RowsetSharedPtr>* tablet_infos,
242
            std::map<TabletInfo, std::shared_ptr<TabletTxnInfo>>* tablet_txn_infos = nullptr);
243
244
    void get_all_related_tablets(std::set<TabletInfo>* tablet_infos);
245
246
    // Get all expired txns and save them in expire_txn_map.
247
    // This is currently called before reporting all tablet info, to avoid iterating txn map for every tablets.
248
    void build_expire_txn_map(std::map<TabletInfo, std::vector<int64_t>>* expire_txn_map);
249
250
    void force_rollback_tablet_related_txns(OlapMeta* meta, TTabletId tablet_id,
251
                                            TabletUid tablet_uid);
252
253
    void get_partition_ids(const TTransactionId transaction_id,
254
                           std::vector<TPartitionId>* partition_ids);
255
256
    void set_txn_related_delete_bitmap(TPartitionId partition_id, TTransactionId transaction_id,
257
                                       TTabletId tablet_id, TabletUid tablet_uid,
258
                                       bool unique_key_merge_on_write,
259
                                       DeleteBitmapPtr delete_bitmap,
260
                                       const RowsetIdUnorderedSet& rowset_ids,
261
                                       std::shared_ptr<PartialUpdateInfo> partial_update_info);
262
    void get_all_commit_tablet_txn_info_by_tablet(
263
            const Tablet& tablet, CommitTabletTxnInfoVec* commit_tablet_txn_info_vec);
264
265
    int64_t get_txn_by_tablet_version(int64_t tablet_id, int64_t version);
266
    void update_tablet_version_txn(int64_t tablet_id, int64_t version, int64_t txn_id);
267
268
    TxnState get_txn_state(TPartitionId partition_id, TTransactionId transaction_id,
269
                           TTabletId tablet_id, TabletUid tablet_uid);
270
271
    void remove_txn_tablet_info(TPartitionId partition_id, TTransactionId transaction_id,
272
                                TTabletId tablet_id, TabletUid tablet_uid);
273
274
private:
275
    using TxnKey = std::pair<int64_t, int64_t>; // partition_id, transaction_id;
276
277
    // Implement TxnKey hash function to support TxnKey as a key for `unordered_map`.
278
    struct TxnKeyHash {
279
        template <typename T, typename U>
280
2.98k
        size_t operator()(const std::pair<T, U>& e) const {
281
2.98k
            return std::hash<T>()(e.first) ^ std::hash<U>()(e.second);
282
2.98k
        }
283
    };
284
285
    // Implement TxnKey equal function to support TxnKey as a key for `unordered_map`.
286
    struct TxnKeyEqual {
287
        template <class T, typename U>
288
2.63k
        bool operator()(const std::pair<T, U>& l, const std::pair<T, U>& r) const {
289
2.63k
            return l.first == r.first && l.second == r.second;
290
2.63k
        }
291
    };
292
293
    using txn_tablet_map_t =
294
            std::unordered_map<TxnKey, std::map<TabletInfo, std::shared_ptr<TabletTxnInfo>>,
295
                               TxnKeyHash, TxnKeyEqual>;
296
    using txn_partition_map_t = std::unordered_map<int64_t, std::unordered_set<int64_t>>;
297
    std::shared_mutex& _get_txn_map_lock(TTransactionId transactionId);
298
299
    txn_tablet_map_t& _get_txn_tablet_map(TTransactionId transactionId);
300
301
    txn_partition_map_t& _get_txn_partition_map(TTransactionId transactionId);
302
303
    inline std::shared_mutex& _get_txn_lock(TTransactionId transactionId);
304
305
    // Insert or remove (transaction_id, partition_id) from _txn_partition_map
306
    // get _txn_map_lock before calling.
307
    void _insert_txn_partition_map_unlocked(int64_t transaction_id, int64_t partition_id);
308
    void _clear_txn_partition_map_unlocked(int64_t transaction_id, int64_t partition_id);
309
310
    void _remove_txn_tablet_info_unlocked(TPartitionId partition_id, TTransactionId transaction_id,
311
                                          TTabletId tablet_id, TabletUid tablet_uid,
312
                                          std::lock_guard<std::shared_mutex>& txn_lock,
313
                                          std::lock_guard<std::shared_mutex>& wrlock);
314
315
    class TabletVersionCache : public LRUCachePolicy {
316
    public:
317
        TabletVersionCache(size_t capacity)
318
703
                : LRUCachePolicy(CachePolicy::CacheType::TABLET_VERSION_CACHE, capacity,
319
703
                                 LRUCacheType::NUMBER, /*sweeptime*/ -1,
320
703
                                 /*num_shards*/ 32,
321
703
                                 /*element_count_capacity*/ 0, /*enable_prune*/ false,
322
703
                                 /*is_lru_k*/ false) {}
323
    };
324
325
private:
326
    StorageEngine& _engine;
327
328
    const int32_t _txn_map_shard_size;
329
330
    const int32_t _txn_shard_size;
331
332
    // _txn_map_locks[i] protect _txn_tablet_maps[i], i=0,1,2...,and i < _txn_map_shard_size
333
    txn_tablet_map_t* _txn_tablet_maps = nullptr;
334
    // transaction_id -> corresponding partition ids
335
    // This is mainly for the clear txn task received from FE, which may only has transaction id,
336
    // so we need this map to find out which partitions are corresponding to a transaction id.
337
    // The _txn_partition_maps[i] should be constructed/deconstructed/modified alongside with '_txn_tablet_maps[i]'
338
    txn_partition_map_t* _txn_partition_maps = nullptr;
339
340
    std::shared_mutex* _txn_map_locks = nullptr;
341
342
    std::shared_mutex* _txn_mutex = nullptr;
343
344
    std::unique_ptr<TabletVersionCache> _tablet_version_cache;
345
    DISALLOW_COPY_AND_ASSIGN(TxnManager);
346
}; // TxnManager
347
348
2.34k
inline std::shared_mutex& TxnManager::_get_txn_map_lock(TTransactionId transactionId) {
349
2.34k
    return _txn_map_locks[transactionId & (_txn_map_shard_size - 1)];
350
2.34k
}
351
352
2.34k
inline TxnManager::txn_tablet_map_t& TxnManager::_get_txn_tablet_map(TTransactionId transactionId) {
353
2.34k
    return _txn_tablet_maps[transactionId & (_txn_map_shard_size - 1)];
354
2.34k
}
355
356
inline TxnManager::txn_partition_map_t& TxnManager::_get_txn_partition_map(
357
1.18k
        TTransactionId transactionId) {
358
1.18k
    return _txn_partition_maps[transactionId & (_txn_map_shard_size - 1)];
359
1.18k
}
360
361
1.38k
inline std::shared_mutex& TxnManager::_get_txn_lock(TTransactionId transactionId) {
362
1.38k
    return _txn_mutex[transactionId & (_txn_shard_size - 1)];
363
1.38k
}
364
365
} // namespace doris