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 |