be/src/cloud/cloud_rowset_builder.cpp
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 | | #include "cloud/cloud_rowset_builder.h" |
19 | | |
20 | | #include "cloud/cloud_meta_mgr.h" |
21 | | #include "cloud/cloud_storage_engine.h" |
22 | | #include "cloud/cloud_tablet.h" |
23 | | #include "cloud/cloud_tablet_mgr.h" |
24 | | #include "storage/rowset/group_rowset_writer.h" |
25 | | #include "storage/rowset/rowset_factory.h" |
26 | | #include "storage/rowset/rowset_writer_context.h" |
27 | | #include "storage/storage_policy.h" |
28 | | #include "storage/tablet_info.h" |
29 | | |
30 | | namespace doris { |
31 | | using namespace ErrorCode; |
32 | | |
33 | | CloudRowsetBuilder::CloudRowsetBuilder(CloudStorageEngine& engine, const WriteRequest& req, |
34 | | RuntimeProfile* profile) |
35 | 266k | : BaseRowsetBuilder(req, profile), _engine(engine) {} |
36 | | |
37 | | CloudGroupRowsetBuilder::CloudGroupRowsetBuilder(CloudStorageEngine& engine, |
38 | | const WriteRequest& group_build_req, |
39 | | const WriteRequest& sub_data_req, |
40 | | const WriteRequest& sub_row_binlog_req, |
41 | | RuntimeProfile* profile) |
42 | 1 | : CloudRowsetBuilder(engine, group_build_req, profile) { |
43 | 1 | DCHECK(group_build_req.write_req_type == WriteRequestType::GROUP && |
44 | 1 | sub_data_req.write_req_type == WriteRequestType::DATA && |
45 | 1 | sub_row_binlog_req.write_req_type == WriteRequestType::ROW_BINLOG); |
46 | 1 | _data_builder = std::make_shared<CloudRowsetBuilder>(engine, sub_data_req, profile); |
47 | 1 | _row_binlog_builder = std::make_shared<CloudRowsetBuilder>(engine, sub_row_binlog_req, profile); |
48 | 1 | } |
49 | | |
50 | 266k | CloudRowsetBuilder::~CloudRowsetBuilder() { |
51 | | // Clear file cache immediately when load fails |
52 | 266k | if (_is_init && _rowset != nullptr && _rowset->rowset_meta()->rowset_state() == PREPARED) { |
53 | 0 | _rowset->clear_cache(); |
54 | 0 | } |
55 | 266k | } |
56 | | |
57 | 2 | Status CloudRowsetBuilder::init() { |
58 | 2 | _tablet = DORIS_TRY(_engine.get_tablet(_req.tablet_id)); |
59 | | |
60 | 2 | std::shared_ptr<MowContext> mow_context; |
61 | 2 | if (_tablet->enable_unique_key_merge_on_write() && is_data_builder()) { |
62 | 1 | if (config::cloud_mow_sync_rowsets_when_load_txn_begin) { |
63 | 0 | auto st = std::static_pointer_cast<CloudTablet>(_tablet)->sync_rowsets(); |
64 | | // sync_rowsets will return INVALID_TABLET_STATE when tablet is under alter |
65 | 0 | if (!st.ok() && !st.is<ErrorCode::INVALID_TABLET_STATE>()) { |
66 | 0 | return st; |
67 | 0 | } |
68 | 0 | } |
69 | 1 | RETURN_IF_ERROR(init_mow_context(mow_context)); |
70 | 1 | } else if (_req.write_req_type == WriteRequestType::ROW_BINLOG) { |
71 | | // Row binlog tablets use txn_delete_bitmap_cache for local make-visible. |
72 | | // The real binlog delete bitmap is derived when the base tablet calculates delete bitmap. |
73 | 1 | _delete_bitmap = std::make_shared<DeleteBitmap>(_req.tablet_id); |
74 | 1 | } |
75 | 2 | RETURN_IF_ERROR(check_tablet_version_count()); |
76 | | |
77 | 2 | using namespace std::chrono; |
78 | 2 | std::static_pointer_cast<CloudTablet>(_tablet)->last_load_time_ms = |
79 | 2 | duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count(); |
80 | | |
81 | | // build tablet schema in request level |
82 | 2 | RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(), |
83 | 2 | *_tablet->tablet_schema())); |
84 | | |
85 | 2 | RowsetWriterContext context; |
86 | 2 | context.txn_id = _req.txn_id; |
87 | 2 | context.txn_expiration = _req.txn_expiration; |
88 | 2 | context.load_id = _req.load_id; |
89 | 2 | context.db_id = _req.table_schema_param->db_id(); |
90 | 2 | context.table_id = _req.table_schema_param->table_id(); |
91 | 2 | context.rowset_state = PREPARED; |
92 | 2 | context.segments_overlap = OVERLAPPING; |
93 | 2 | context.tablet_schema = _tablet_schema; |
94 | 2 | context.newest_write_timestamp = UnixSeconds(); |
95 | 2 | context.tablet_id = _req.tablet_id; |
96 | 2 | context.tablet_schema_hash = _req.schema_hash; |
97 | 2 | context.index_id = _req.index_id; |
98 | 2 | context.tablet = _tablet; |
99 | 2 | if (_req.write_req_type == WriteRequestType::ROW_BINLOG || !_attach_rowset_ids.empty()) { |
100 | 2 | context.enable_segcompaction = false; |
101 | 2 | } |
102 | 2 | context.write_type = DataWriteType::TYPE_DIRECT; |
103 | 2 | context.mow_context = mow_context; |
104 | 2 | context.write_file_cache = _req.write_file_cache; |
105 | 2 | context.partial_update_info = _partial_update_info; |
106 | 2 | context.write_binlog_opt().enable = _req.write_req_type == WriteRequestType::ROW_BINLOG; |
107 | 2 | context.file_cache_ttl_sec = _tablet->ttl_seconds(); |
108 | 2 | context.storage_resource = _engine.get_storage_resource(_req.storage_vault_id); |
109 | 2 | if (!context.storage_resource) { |
110 | 0 | return Status::InternalError("vault id not found, maybe not sync, vault id {}", |
111 | 0 | _req.storage_vault_id); |
112 | 0 | } |
113 | | |
114 | 2 | _rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false)); |
115 | 2 | _rowset_id = context.rowset_id; |
116 | | |
117 | 2 | _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token(); |
118 | | |
119 | 2 | if (!_skip_writing_rowset_metadata) { |
120 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().prepare_rowset(*_rowset_writer->rowset_meta(), "", |
121 | 0 | _tablet->table_id())); |
122 | 0 | } |
123 | | |
124 | 2 | _is_init = true; |
125 | 2 | return Status::OK(); |
126 | 2 | } |
127 | | |
128 | 1 | Status CloudGroupRowsetBuilder::init() { |
129 | 1 | RETURN_IF_ERROR(_row_binlog_builder->init()); |
130 | 1 | RETURN_IF_ERROR( |
131 | 1 | _data_builder->attach_pending_rs_guard_to_txn(_row_binlog_builder->rowset_id())); |
132 | 1 | RETURN_IF_ERROR(_data_builder->init()); |
133 | 1 | _tablet = _data_builder->tablet_sptr(); |
134 | | |
135 | 1 | std::unique_ptr<GroupRowsetWriter> group_writer; |
136 | 1 | RETURN_IF_ERROR(RowsetFactory::create_empty_group_rowset_writer(&group_writer)); |
137 | 1 | group_writer->set_data_writer(_data_builder->rowset_writer()); |
138 | 1 | group_writer->set_row_binlog_writer(_row_binlog_builder->rowset_writer()); |
139 | | |
140 | 1 | { |
141 | 1 | const auto& data_ctx = _data_builder->rowset_writer()->context(); |
142 | 1 | auto& binlog_ctx = |
143 | 1 | const_cast<RowsetWriterContext&>(_row_binlog_builder->rowset_writer()->context()); |
144 | 1 | auto& cfg = binlog_ctx.write_binlog_opt().write_binlog_config(); |
145 | 1 | cfg.source.tablet_schema = data_ctx.tablet_schema; |
146 | 1 | cfg.source.partial_update_info = data_ctx.partial_update_info; |
147 | 1 | cfg.source.mow_context = data_ctx.mow_context; |
148 | 1 | cfg.source.is_transient_rowset_writer = data_ctx.is_transient_rowset_writer; |
149 | 1 | cfg.source.source_write_type = data_ctx.write_type; |
150 | 1 | cfg.source.base_tablet = _data_builder->tablet_sptr(); |
151 | 1 | } |
152 | | |
153 | 1 | _rowset_writer = std::move(group_writer); |
154 | 1 | _is_init = true; |
155 | 1 | return Status::OK(); |
156 | 1 | } |
157 | | |
158 | 1 | Status CloudGroupRowsetBuilder::build_rowset() { |
159 | 1 | RETURN_IF_ERROR(_row_binlog_builder->build_rowset()); |
160 | 1 | return _data_builder->build_rowset(); |
161 | 1 | } |
162 | | |
163 | 0 | Status CloudGroupRowsetBuilder::submit_calc_delete_bitmap_task() { |
164 | 0 | return _data_builder->submit_calc_delete_bitmap_task(); |
165 | 0 | } |
166 | | |
167 | 0 | Status CloudGroupRowsetBuilder::wait_calc_delete_bitmap() { |
168 | 0 | return _data_builder->wait_calc_delete_bitmap(); |
169 | 0 | } |
170 | | |
171 | 0 | void CloudGroupRowsetBuilder::update_tablet_stats() { |
172 | 0 | _data_builder->update_tablet_stats(); |
173 | 0 | _row_binlog_builder->update_tablet_stats(); |
174 | 0 | } |
175 | | |
176 | 0 | Status CloudGroupRowsetBuilder::commit_rowset(const std::string& job_id, int64_t table_id) { |
177 | 0 | return _engine.meta_mgr().commit_rowset(*_data_builder->rowset_meta(), job_id, table_id, |
178 | 0 | nullptr, _row_binlog_builder->rowset_meta().get()); |
179 | 0 | } |
180 | | |
181 | 0 | Status CloudGroupRowsetBuilder::set_txn_related_info() { |
182 | 0 | RowBinlogTxnInfo attach_row_binlog; |
183 | 0 | attach_row_binlog.rowset = _row_binlog_builder->rowset(); |
184 | 0 | attach_row_binlog.tablet = _row_binlog_builder->tablet_sptr(); |
185 | 0 | if (_data_builder->tablet()->enable_unique_key_merge_on_write()) { |
186 | 0 | attach_row_binlog.delete_bitmap = |
187 | 0 | std::make_shared<DeleteBitmap>(_row_binlog_builder->tablet()->tablet_id()); |
188 | 0 | } |
189 | 0 | RETURN_IF_ERROR(_data_builder->attach_row_binlog_to_txn(attach_row_binlog)); |
190 | 0 | RETURN_IF_ERROR(_data_builder->set_txn_related_info()); |
191 | 0 | return _row_binlog_builder->set_txn_related_info(); |
192 | 0 | } |
193 | | |
194 | 1 | void CloudGroupRowsetBuilder::set_skip_writing_rowset_metadata(bool skip) { |
195 | 1 | _data_builder->set_skip_writing_rowset_metadata(skip); |
196 | 1 | _row_binlog_builder->set_skip_writing_rowset_metadata(skip); |
197 | 1 | } |
198 | | |
199 | 173k | Status CloudRowsetBuilder::check_tablet_version_count() { |
200 | 173k | int64_t version_count = cloud_tablet()->fetch_add_approximate_num_rowsets(0); |
201 | 173k | DBUG_EXECUTE_IF("RowsetBuilder.check_tablet_version_count.too_many_version", |
202 | 173k | { version_count = INT_MAX; }); |
203 | | // TODO(plat1ko): load backoff algorithm |
204 | 173k | int32_t max_version_config = cloud_tablet()->max_version_config(); |
205 | 173k | if (version_count > max_version_config) { |
206 | 0 | return Status::Error<TOO_MANY_VERSION>( |
207 | 0 | "failed to init rowset builder. version count: {}, exceed limit: {}, " |
208 | 0 | "tablet: {}. Please reduce the frequency of loading data or adjust the " |
209 | 0 | "max_tablet_version_num or time_series_max_tablet_version_numin be.conf to a " |
210 | 0 | "larger value.", |
211 | 0 | version_count, max_version_config, _tablet->tablet_id()); |
212 | 0 | } |
213 | 173k | return Status::OK(); |
214 | 173k | } |
215 | | |
216 | 174k | void CloudRowsetBuilder::update_tablet_stats() { |
217 | 174k | auto* tablet = cloud_tablet(); |
218 | 174k | DCHECK(tablet); |
219 | 174k | DCHECK(_rowset); |
220 | 174k | tablet->fetch_add_approximate_num_rowsets(1); |
221 | 174k | tablet->fetch_add_approximate_num_segments(_rowset->num_segments()); |
222 | 174k | tablet->fetch_add_approximate_num_rows(_rowset->num_rows()); |
223 | 174k | tablet->fetch_add_approximate_data_size(_rowset->total_disk_size()); |
224 | 174k | tablet->fetch_add_approximate_cumu_num_rowsets(1); |
225 | 174k | tablet->fetch_add_approximate_cumu_num_deltas(_rowset->num_segments()); |
226 | 174k | tablet->write_count.fetch_add(1, std::memory_order_relaxed); |
227 | 174k | } |
228 | | |
229 | 521k | CloudTablet* CloudRowsetBuilder::cloud_tablet() { |
230 | 521k | return static_cast<CloudTablet*>(_tablet.get()); |
231 | 521k | } |
232 | | |
233 | 87.9k | const RowsetMetaSharedPtr& CloudRowsetBuilder::rowset_meta() { |
234 | 87.9k | return _rowset_writer->rowset_meta(); |
235 | 87.9k | } |
236 | | |
237 | 0 | Status CloudRowsetBuilder::commit_rowset(const std::string& job_id, int64_t table_id) { |
238 | 0 | return _engine.meta_mgr().commit_rowset(*rowset_meta(), job_id, table_id); |
239 | 0 | } |
240 | | |
241 | 0 | Status CloudRowsetBuilder::set_txn_related_info() { |
242 | 0 | if (_tablet->enable_unique_key_merge_on_write() || _tablet->is_row_binlog_tablet()) { |
243 | | // For empty rowsets when skip_writing_empty_rowset_metadata=true, |
244 | | // store only a lightweight marker instead of full rowset info. |
245 | | // This allows CalcDeleteBitmapTask to detect and skip gracefully, |
246 | | // while using minimal memory (~16 bytes per entry). |
247 | 0 | if (_skip_writing_rowset_metadata) { |
248 | 0 | _engine.txn_delete_bitmap_cache().mark_empty_rowset(_req.txn_id, _tablet->tablet_id(), |
249 | 0 | _req.txn_expiration); |
250 | 0 | return Status::OK(); |
251 | 0 | } |
252 | 0 | if (config::enable_merge_on_write_correctness_check && |
253 | 0 | _tablet->enable_unique_key_merge_on_write() && _rowset->num_rows() != 0) { |
254 | 0 | auto st = _tablet->check_delete_bitmap_correctness( |
255 | 0 | _delete_bitmap, _rowset->end_version() - 1, _req.txn_id, *_rowset_ids); |
256 | 0 | if (!st.ok()) { |
257 | 0 | LOG(WARNING) << fmt::format( |
258 | 0 | "[tablet_id:{}][txn_id:{}][load_id:{}][partition_id:{}] " |
259 | 0 | "delete bitmap correctness check failed in commit phase!", |
260 | 0 | _req.tablet_id, _req.txn_id, UniqueId(_req.load_id).to_string(), |
261 | 0 | _req.partition_id); |
262 | 0 | return st; |
263 | 0 | } |
264 | 0 | } |
265 | 0 | _engine.txn_delete_bitmap_cache().set_tablet_txn_info( |
266 | 0 | _req.txn_id, _tablet->tablet_id(), _delete_bitmap, *_rowset_ids, _rowset, |
267 | 0 | _req.txn_expiration, _partial_update_info, _attach_row_binlog); |
268 | 0 | } else { |
269 | | // TSO-enabled rowsets must become visible from MS rowset meta. |
270 | 0 | if (config::enable_cloud_make_rs_visible_on_be && !_tablet_schema->is_tso_enabled()) { |
271 | 0 | if (_skip_writing_rowset_metadata) { |
272 | 0 | _engine.committed_rs_mgr().mark_empty_rowset(_req.txn_id, _tablet->tablet_id(), |
273 | 0 | _req.txn_expiration); |
274 | 0 | } else { |
275 | 0 | _engine.meta_mgr().cache_committed_rowset(rowset_meta(), _req.txn_expiration); |
276 | 0 | } |
277 | 0 | } |
278 | 0 | } |
279 | 0 | return Status::OK(); |
280 | 0 | } |
281 | | } // namespace doris |