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