/root/doris/be/src/olap/schema_change.cpp
Line | Count | Source (jump to first uncovered line) |
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 "olap/schema_change.h" |
19 | | |
20 | | #include <gen_cpp/olap_file.pb.h> |
21 | | #include <glog/logging.h> |
22 | | #include <thrift/protocol/TDebugProtocol.h> |
23 | | |
24 | | #include <algorithm> |
25 | | #include <exception> |
26 | | #include <map> |
27 | | #include <memory> |
28 | | #include <mutex> |
29 | | #include <roaring/roaring.hh> |
30 | | #include <tuple> |
31 | | #include <utility> |
32 | | |
33 | | #include "agent/be_exec_version_manager.h" |
34 | | #include "cloud/cloud_schema_change_job.h" |
35 | | #include "cloud/config.h" |
36 | | #include "common/consts.h" |
37 | | #include "common/logging.h" |
38 | | #include "common/signal_handler.h" |
39 | | #include "common/status.h" |
40 | | #include "exec/schema_scanner/schema_metadata_name_ids_scanner.h" |
41 | | #include "gutil/integral_types.h" |
42 | | #include "gutil/strings/numbers.h" |
43 | | #include "io/fs/file_system.h" |
44 | | #include "io/io_common.h" |
45 | | #include "olap/base_tablet.h" |
46 | | #include "olap/data_dir.h" |
47 | | #include "olap/delete_handler.h" |
48 | | #include "olap/field.h" |
49 | | #include "olap/iterators.h" |
50 | | #include "olap/merger.h" |
51 | | #include "olap/olap_common.h" |
52 | | #include "olap/olap_define.h" |
53 | | #include "olap/rowset/beta_rowset.h" |
54 | | #include "olap/rowset/pending_rowset_helper.h" |
55 | | #include "olap/rowset/rowset_meta.h" |
56 | | #include "olap/rowset/rowset_reader_context.h" |
57 | | #include "olap/rowset/rowset_writer_context.h" |
58 | | #include "olap/rowset/segment_v2/column_reader.h" |
59 | | #include "olap/rowset/segment_v2/inverted_index_desc.h" |
60 | | #include "olap/rowset/segment_v2/inverted_index_writer.h" |
61 | | #include "olap/rowset/segment_v2/segment.h" |
62 | | #include "olap/schema.h" |
63 | | #include "olap/segment_loader.h" |
64 | | #include "olap/storage_engine.h" |
65 | | #include "olap/tablet.h" |
66 | | #include "olap/tablet_fwd.h" |
67 | | #include "olap/tablet_manager.h" |
68 | | #include "olap/tablet_meta.h" |
69 | | #include "olap/tablet_schema.h" |
70 | | #include "olap/types.h" |
71 | | #include "olap/utils.h" |
72 | | #include "olap/wrapper_field.h" |
73 | | #include "runtime/exec_env.h" |
74 | | #include "runtime/memory/mem_tracker.h" |
75 | | #include "runtime/runtime_state.h" |
76 | | #include "util/debug_points.h" |
77 | | #include "util/defer_op.h" |
78 | | #include "util/trace.h" |
79 | | #include "vec/aggregate_functions/aggregate_function.h" |
80 | | #include "vec/aggregate_functions/aggregate_function_reader.h" |
81 | | #include "vec/columns/column.h" |
82 | | #include "vec/columns/column_nullable.h" |
83 | | #include "vec/common/assert_cast.h" |
84 | | #include "vec/common/schema_util.h" |
85 | | #include "vec/core/block.h" |
86 | | #include "vec/core/column_with_type_and_name.h" |
87 | | #include "vec/exprs/vexpr.h" |
88 | | #include "vec/exprs/vexpr_context.h" |
89 | | #include "vec/olap/olap_data_convertor.h" |
90 | | |
91 | | namespace doris { |
92 | | class CollectionValue; |
93 | | |
94 | | using namespace ErrorCode; |
95 | | |
96 | | constexpr int ALTER_TABLE_BATCH_SIZE = 4064; |
97 | | |
98 | | class MultiBlockMerger { |
99 | | public: |
100 | 0 | MultiBlockMerger(BaseTabletSPtr tablet) : _tablet(tablet), _cmp(*tablet) {} |
101 | | |
102 | | Status merge(const std::vector<std::unique_ptr<vectorized::Block>>& blocks, |
103 | 0 | RowsetWriter* rowset_writer, uint64_t* merged_rows) { |
104 | 0 | int rows = 0; |
105 | 0 | for (const auto& block : blocks) { |
106 | 0 | rows += block->rows(); |
107 | 0 | } |
108 | 0 | if (!rows) { |
109 | 0 | return Status::OK(); |
110 | 0 | } |
111 | | |
112 | 0 | std::vector<RowRef> row_refs; |
113 | 0 | row_refs.reserve(rows); |
114 | 0 | for (const auto& block : blocks) { |
115 | 0 | for (uint16_t i = 0; i < block->rows(); i++) { |
116 | 0 | row_refs.emplace_back(block.get(), i); |
117 | 0 | } |
118 | 0 | } |
119 | | // TODO: try to use pdqsort to replace std::sort |
120 | | // The block version is incremental. |
121 | 0 | std::stable_sort(row_refs.begin(), row_refs.end(), _cmp); |
122 | |
|
123 | 0 | auto finalized_block = _tablet->tablet_schema()->create_block(); |
124 | 0 | int columns = finalized_block.columns(); |
125 | 0 | *merged_rows += rows; |
126 | |
|
127 | 0 | if (_tablet->keys_type() == KeysType::AGG_KEYS) { |
128 | 0 | auto tablet_schema = _tablet->tablet_schema(); |
129 | 0 | int key_number = _tablet->num_key_columns(); |
130 | |
|
131 | 0 | std::vector<vectorized::AggregateFunctionPtr> agg_functions; |
132 | 0 | std::vector<vectorized::AggregateDataPtr> agg_places; |
133 | |
|
134 | 0 | for (int i = key_number; i < columns; i++) { |
135 | 0 | try { |
136 | 0 | vectorized::AggregateFunctionPtr function = |
137 | 0 | tablet_schema->column(i).get_aggregate_function( |
138 | 0 | vectorized::AGG_LOAD_SUFFIX, |
139 | 0 | tablet_schema->column(i).get_be_exec_version()); |
140 | 0 | if (!function) { |
141 | 0 | return Status::InternalError( |
142 | 0 | "could not find aggregate function on column {}, aggregation={}", |
143 | 0 | tablet_schema->column(i).name(), |
144 | 0 | tablet_schema->column(i).aggregation()); |
145 | 0 | } |
146 | 0 | agg_functions.push_back(function); |
147 | | // create aggregate data |
148 | 0 | auto* place = new char[function->size_of_data()]; |
149 | 0 | function->create(place); |
150 | 0 | agg_places.push_back(place); |
151 | 0 | } catch (...) { |
152 | 0 | for (int j = 0; j < i - key_number; ++j) { |
153 | 0 | agg_functions[j]->destroy(agg_places[j]); |
154 | 0 | delete[] agg_places[j]; |
155 | 0 | } |
156 | 0 | throw; |
157 | 0 | } |
158 | 0 | } |
159 | | |
160 | 0 | DEFER({ |
161 | 0 | for (int i = 0; i < columns - key_number; i++) { |
162 | 0 | agg_functions[i]->destroy(agg_places[i]); |
163 | 0 | delete[] agg_places[i]; |
164 | 0 | } |
165 | 0 | }); |
166 | |
|
167 | 0 | for (int i = 0; i < rows; i++) { |
168 | 0 | auto row_ref = row_refs[i]; |
169 | 0 | for (int j = key_number; j < columns; j++) { |
170 | 0 | const auto* column_ptr = row_ref.get_column(j).get(); |
171 | 0 | agg_functions[j - key_number]->add( |
172 | 0 | agg_places[j - key_number], |
173 | 0 | const_cast<const vectorized::IColumn**>(&column_ptr), row_ref.position, |
174 | 0 | &_arena); |
175 | 0 | } |
176 | |
|
177 | 0 | if (i == rows - 1 || _cmp.compare(row_refs[i], row_refs[i + 1])) { |
178 | 0 | for (int j = 0; j < key_number; j++) { |
179 | 0 | finalized_block.get_by_position(j).column->assume_mutable()->insert_from( |
180 | 0 | *row_ref.get_column(j), row_ref.position); |
181 | 0 | } |
182 | |
|
183 | 0 | for (int j = key_number; j < columns; j++) { |
184 | 0 | agg_functions[j - key_number]->insert_result_into( |
185 | 0 | agg_places[j - key_number], |
186 | 0 | finalized_block.get_by_position(j).column->assume_mutable_ref()); |
187 | 0 | agg_functions[j - key_number]->reset(agg_places[j - key_number]); |
188 | 0 | } |
189 | |
|
190 | 0 | if (i == rows - 1 || finalized_block.rows() == ALTER_TABLE_BATCH_SIZE) { |
191 | 0 | *merged_rows -= finalized_block.rows(); |
192 | 0 | RETURN_IF_ERROR(rowset_writer->add_block(&finalized_block)); |
193 | 0 | finalized_block.clear_column_data(); |
194 | 0 | } |
195 | 0 | } |
196 | 0 | } |
197 | 0 | } else { |
198 | 0 | std::vector<RowRef> pushed_row_refs; |
199 | 0 | if (_tablet->keys_type() == KeysType::DUP_KEYS) { |
200 | 0 | std::swap(pushed_row_refs, row_refs); |
201 | 0 | } else if (_tablet->keys_type() == KeysType::UNIQUE_KEYS) { |
202 | 0 | for (int i = 0; i < rows; i++) { |
203 | 0 | if (i == rows - 1 || _cmp.compare(row_refs[i], row_refs[i + 1])) { |
204 | 0 | pushed_row_refs.push_back(row_refs[i]); |
205 | 0 | } |
206 | 0 | } |
207 | 0 | if (!_tablet->tablet_schema()->cluster_key_uids().empty()) { |
208 | 0 | std::vector<uint32_t> ids; |
209 | 0 | for (const auto& cid : _tablet->tablet_schema()->cluster_key_uids()) { |
210 | 0 | auto index = _tablet->tablet_schema()->field_index(cid); |
211 | 0 | if (index == -1) { |
212 | 0 | return Status::InternalError( |
213 | 0 | "could not find cluster key column with unique_id=" + |
214 | 0 | std::to_string(cid) + " in tablet schema"); |
215 | 0 | } |
216 | 0 | ids.push_back(index); |
217 | 0 | } |
218 | | // sort by cluster key |
219 | 0 | std::stable_sort(pushed_row_refs.begin(), pushed_row_refs.end(), |
220 | 0 | ClusterKeyRowRefComparator(ids)); |
221 | 0 | } |
222 | 0 | } |
223 | | |
224 | | // update real inserted row number |
225 | 0 | rows = pushed_row_refs.size(); |
226 | 0 | *merged_rows -= rows; |
227 | |
|
228 | 0 | for (int i = 0; i < rows; i += ALTER_TABLE_BATCH_SIZE) { |
229 | 0 | int limit = std::min(ALTER_TABLE_BATCH_SIZE, rows - i); |
230 | |
|
231 | 0 | for (int idx = 0; idx < columns; idx++) { |
232 | 0 | auto column = finalized_block.get_by_position(idx).column->assume_mutable(); |
233 | |
|
234 | 0 | for (int j = 0; j < limit; j++) { |
235 | 0 | auto row_ref = pushed_row_refs[i + j]; |
236 | 0 | column->insert_from(*row_ref.get_column(idx), row_ref.position); |
237 | 0 | } |
238 | 0 | } |
239 | 0 | RETURN_IF_ERROR(rowset_writer->add_block(&finalized_block)); |
240 | 0 | finalized_block.clear_column_data(); |
241 | 0 | } |
242 | 0 | } |
243 | | |
244 | 0 | RETURN_IF_ERROR(rowset_writer->flush()); |
245 | 0 | return Status::OK(); |
246 | 0 | } |
247 | | |
248 | | private: |
249 | | struct RowRef { |
250 | | RowRef(vectorized::Block* block_, uint16_t position_) |
251 | 0 | : block(block_), position(position_) {} |
252 | 0 | vectorized::ColumnPtr get_column(int index) const { |
253 | 0 | return block->get_by_position(index).column; |
254 | 0 | } |
255 | | const vectorized::Block* block; |
256 | | uint16_t position; |
257 | | }; |
258 | | |
259 | | struct RowRefComparator { |
260 | 0 | RowRefComparator(const BaseTablet& tablet) : _num_columns(tablet.num_key_columns()) {} |
261 | | |
262 | 0 | int compare(const RowRef& lhs, const RowRef& rhs) const { |
263 | | // Notice: does not compare sequence column for mow table |
264 | | // read from rowsets with delete bitmap, so there should be no duplicated keys |
265 | 0 | return lhs.block->compare_at(lhs.position, rhs.position, _num_columns, *rhs.block, -1); |
266 | 0 | } |
267 | | |
268 | 0 | bool operator()(const RowRef& lhs, const RowRef& rhs) const { |
269 | 0 | return compare(lhs, rhs) < 0; |
270 | 0 | } |
271 | | |
272 | | const size_t _num_columns; |
273 | | }; |
274 | | |
275 | | struct ClusterKeyRowRefComparator { |
276 | 0 | ClusterKeyRowRefComparator(std::vector<uint32_t> columns) : _columns(columns) {} |
277 | | |
278 | 0 | int compare(const RowRef& lhs, const RowRef& rhs) const { |
279 | 0 | return lhs.block->compare_at(lhs.position, rhs.position, &_columns, *rhs.block, -1); |
280 | 0 | } |
281 | | |
282 | 0 | bool operator()(const RowRef& lhs, const RowRef& rhs) const { |
283 | 0 | return compare(lhs, rhs) < 0; |
284 | 0 | } |
285 | | |
286 | | const std::vector<uint32_t> _columns; |
287 | | }; |
288 | | |
289 | | BaseTabletSPtr _tablet; |
290 | | RowRefComparator _cmp; |
291 | | vectorized::Arena _arena; |
292 | | }; |
293 | | |
294 | | BlockChanger::BlockChanger(TabletSchemaSPtr tablet_schema, DescriptorTbl desc_tbl) |
295 | 0 | : _desc_tbl(std::move(desc_tbl)) { |
296 | 0 | _schema_mapping.resize(tablet_schema->num_columns()); |
297 | 0 | } |
298 | | |
299 | 0 | BlockChanger::~BlockChanger() { |
300 | 0 | for (auto it = _schema_mapping.begin(); it != _schema_mapping.end(); ++it) { |
301 | 0 | SAFE_DELETE(it->default_value); |
302 | 0 | } |
303 | 0 | _schema_mapping.clear(); |
304 | 0 | } |
305 | | |
306 | 0 | ColumnMapping* BlockChanger::get_mutable_column_mapping(size_t column_index) { |
307 | 0 | if (column_index >= _schema_mapping.size()) { |
308 | 0 | return nullptr; |
309 | 0 | } |
310 | | |
311 | 0 | return &(_schema_mapping[column_index]); |
312 | 0 | } |
313 | | |
314 | | Status BlockChanger::change_block(vectorized::Block* ref_block, |
315 | 0 | vectorized::Block* new_block) const { |
316 | 0 | std::unique_ptr<RuntimeState> state = RuntimeState::create_unique(); |
317 | 0 | state->set_desc_tbl(&_desc_tbl); |
318 | 0 | state->set_be_exec_version(_fe_compatible_version); |
319 | 0 | RowDescriptor row_desc = |
320 | 0 | RowDescriptor(_desc_tbl.get_tuple_descriptor(_desc_tbl.get_row_tuples()[0]), false); |
321 | |
|
322 | 0 | if (_where_expr != nullptr) { |
323 | 0 | vectorized::VExprContextSPtr ctx = nullptr; |
324 | 0 | RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(*_where_expr, ctx)); |
325 | 0 | RETURN_IF_ERROR(ctx->prepare(state.get(), row_desc)); |
326 | 0 | RETURN_IF_ERROR(ctx->open(state.get())); |
327 | | |
328 | 0 | RETURN_IF_ERROR( |
329 | 0 | vectorized::VExprContext::filter_block(ctx.get(), ref_block, ref_block->columns())); |
330 | 0 | } |
331 | | |
332 | 0 | const int row_num = ref_block->rows(); |
333 | 0 | const int new_schema_cols_num = new_block->columns(); |
334 | | |
335 | | // will be used for swaping ref_block[entry.first] and new_block[entry.second] |
336 | 0 | std::list<std::pair<int, int>> swap_idx_list; |
337 | 0 | for (int idx = 0; idx < new_schema_cols_num; idx++) { |
338 | 0 | auto expr = _schema_mapping[idx].expr; |
339 | 0 | if (expr != nullptr) { |
340 | 0 | vectorized::VExprContextSPtr ctx; |
341 | 0 | RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(*expr, ctx)); |
342 | 0 | RETURN_IF_ERROR(ctx->prepare(state.get(), row_desc)); |
343 | 0 | RETURN_IF_ERROR(ctx->open(state.get())); |
344 | | |
345 | 0 | int result_tmp_column_idx = -1; |
346 | 0 | RETURN_IF_ERROR(ctx->execute(ref_block, &result_tmp_column_idx)); |
347 | 0 | auto& result_tmp_column_def = ref_block->get_by_position(result_tmp_column_idx); |
348 | 0 | if (!result_tmp_column_def.column) { |
349 | 0 | return Status::Error<ErrorCode::INTERNAL_ERROR>( |
350 | 0 | "result column={} is nullptr, input expr={}", result_tmp_column_def.name, |
351 | 0 | apache::thrift::ThriftDebugString(*expr)); |
352 | 0 | } |
353 | 0 | ref_block->replace_by_position_if_const(result_tmp_column_idx); |
354 | |
|
355 | 0 | if (result_tmp_column_def.column->size() != row_num) { |
356 | 0 | return Status::Error<ErrorCode::INTERNAL_ERROR>( |
357 | 0 | "result size invalid, expect={}, real={}; input expr={}, block={}", row_num, |
358 | 0 | result_tmp_column_def.column->size(), |
359 | 0 | apache::thrift::ThriftDebugString(*expr), ref_block->dump_structure()); |
360 | 0 | } |
361 | | |
362 | 0 | if (_type == SCHEMA_CHANGE) { |
363 | | // danger casts (expected to be rejected by upstream caller) may cause data to be null and result in data loss in schema change |
364 | | // for rollup, this check is unecessary, and ref columns are not set in this case, it works on exprs |
365 | | |
366 | | // column_idx in base schema |
367 | 0 | int32_t ref_column_idx = _schema_mapping[idx].ref_column_idx; |
368 | 0 | DCHECK_GE(ref_column_idx, 0); |
369 | 0 | auto& ref_column_def = ref_block->get_by_position(ref_column_idx); |
370 | 0 | RETURN_IF_ERROR( |
371 | 0 | _check_cast_valid(ref_column_def.column, result_tmp_column_def.column)); |
372 | 0 | } |
373 | 0 | swap_idx_list.emplace_back(result_tmp_column_idx, idx); |
374 | 0 | } else if (_schema_mapping[idx].ref_column_idx < 0) { |
375 | | // new column, write default value |
376 | 0 | auto* value = _schema_mapping[idx].default_value; |
377 | 0 | auto column = new_block->get_by_position(idx).column->assume_mutable(); |
378 | 0 | if (value->is_null()) { |
379 | 0 | DCHECK(column->is_nullable()); |
380 | 0 | column->insert_many_defaults(row_num); |
381 | 0 | } else { |
382 | 0 | auto type_info = get_type_info(_schema_mapping[idx].new_column); |
383 | 0 | DefaultValueColumnIterator::insert_default_data(type_info.get(), value->size(), |
384 | 0 | value->ptr(), column, row_num); |
385 | 0 | } |
386 | 0 | } else { |
387 | | // same type, just swap column |
388 | 0 | swap_idx_list.emplace_back(_schema_mapping[idx].ref_column_idx, idx); |
389 | 0 | } |
390 | 0 | } |
391 | | |
392 | 0 | for (auto it : swap_idx_list) { |
393 | 0 | auto& ref_col = ref_block->get_by_position(it.first).column; |
394 | 0 | auto& new_col = new_block->get_by_position(it.second).column; |
395 | |
|
396 | 0 | bool ref_col_nullable = ref_col->is_nullable(); |
397 | 0 | bool new_col_nullable = new_col->is_nullable(); |
398 | |
|
399 | 0 | if (ref_col_nullable != new_col_nullable) { |
400 | | // not nullable to nullable |
401 | 0 | if (new_col_nullable) { |
402 | 0 | auto* new_nullable_col = |
403 | 0 | assert_cast<vectorized::ColumnNullable*>(new_col->assume_mutable().get()); |
404 | |
|
405 | 0 | new_nullable_col->change_nested_column(ref_col); |
406 | 0 | new_nullable_col->get_null_map_data().resize_fill(ref_col->size()); |
407 | 0 | } else { |
408 | | // nullable to not nullable: |
409 | | // suppose column `c_phone` is originally varchar(16) NOT NULL, |
410 | | // then do schema change `alter table test modify column c_phone int not null`, |
411 | | // the cast expr of schema change is `CastExpr(CAST String to Nullable(Int32))`, |
412 | | // so need to handle nullable to not nullable here |
413 | 0 | auto* ref_nullable_col = |
414 | 0 | assert_cast<vectorized::ColumnNullable*>(ref_col->assume_mutable().get()); |
415 | |
|
416 | 0 | new_col = ref_nullable_col->get_nested_column_ptr(); |
417 | 0 | } |
418 | 0 | } else { |
419 | 0 | new_block->get_by_position(it.second).column = |
420 | 0 | ref_block->get_by_position(it.first).column; |
421 | 0 | } |
422 | 0 | } |
423 | 0 | return Status::OK(); |
424 | 0 | } |
425 | | |
426 | | // This check can prevent schema-change from causing data loss after type cast |
427 | | Status BlockChanger::_check_cast_valid(vectorized::ColumnPtr input_column, |
428 | 0 | vectorized::ColumnPtr output_column) { |
429 | 0 | if (input_column->size() != output_column->size()) { |
430 | 0 | return Status::InternalError( |
431 | 0 | "column size is changed, input_column_size={}, output_column_size={}; " |
432 | 0 | "input_column={}", |
433 | 0 | input_column->size(), output_column->size(), input_column->get_name()); |
434 | 0 | } |
435 | 0 | DCHECK_EQ(input_column->size(), output_column->size()) |
436 | 0 | << "length check should have done before calling this function!"; |
437 | |
|
438 | 0 | if (input_column->is_nullable() != output_column->is_nullable()) { |
439 | 0 | if (input_column->is_nullable()) { |
440 | 0 | const auto* ref_null_map = |
441 | 0 | vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column.get()) |
442 | 0 | ->get_null_map_column() |
443 | 0 | .get_data() |
444 | 0 | .data(); |
445 | |
|
446 | 0 | bool is_changed = false; |
447 | 0 | for (size_t i = 0; i < input_column->size(); i++) { |
448 | 0 | is_changed |= ref_null_map[i]; |
449 | 0 | } |
450 | 0 | if (is_changed) { |
451 | 0 | return Status::DataQualityError( |
452 | 0 | "some null data is changed to not null, intput_column={}", |
453 | 0 | input_column->get_name()); |
454 | 0 | } |
455 | 0 | } else { |
456 | 0 | const auto& null_map_column = |
457 | 0 | vectorized::check_and_get_column<vectorized::ColumnNullable>( |
458 | 0 | output_column.get()) |
459 | 0 | ->get_null_map_column(); |
460 | 0 | const auto& nested_column = |
461 | 0 | vectorized::check_and_get_column<vectorized::ColumnNullable>( |
462 | 0 | output_column.get()) |
463 | 0 | ->get_nested_column(); |
464 | 0 | const auto* new_null_map = null_map_column.get_data().data(); |
465 | |
|
466 | 0 | if (null_map_column.size() != output_column->size()) { |
467 | 0 | return Status::InternalError( |
468 | 0 | "null_map_column size mismatch output_column_size, " |
469 | 0 | "null_map_column_size={}, output_column_size={}; input_column={}", |
470 | 0 | null_map_column.size(), output_column->size(), input_column->get_name()); |
471 | 0 | } |
472 | | |
473 | 0 | if (nested_column.size() != output_column->size()) { |
474 | 0 | return Status::InternalError( |
475 | 0 | "nested_column size is changed, nested_column_size={}, " |
476 | 0 | "ouput_column_size={}; input_column={}", |
477 | 0 | nested_column.size(), output_column->size(), input_column->get_name()); |
478 | 0 | } |
479 | | |
480 | 0 | bool is_changed = false; |
481 | 0 | for (size_t i = 0; i < input_column->size(); i++) { |
482 | 0 | is_changed |= new_null_map[i]; |
483 | 0 | } |
484 | 0 | if (is_changed) { |
485 | 0 | return Status::DataQualityError( |
486 | 0 | "some not null data is changed to null, intput_column={}", |
487 | 0 | input_column->get_name()); |
488 | 0 | } |
489 | 0 | } |
490 | 0 | } |
491 | | |
492 | 0 | if (input_column->is_nullable() && output_column->is_nullable()) { |
493 | 0 | const auto* ref_null_map = |
494 | 0 | vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column.get()) |
495 | 0 | ->get_null_map_column() |
496 | 0 | .get_data() |
497 | 0 | .data(); |
498 | 0 | const auto* new_null_map = |
499 | 0 | vectorized::check_and_get_column<vectorized::ColumnNullable>(output_column.get()) |
500 | 0 | ->get_null_map_column() |
501 | 0 | .get_data() |
502 | 0 | .data(); |
503 | |
|
504 | 0 | bool is_changed = false; |
505 | 0 | for (size_t i = 0; i < input_column->size(); i++) { |
506 | 0 | is_changed |= (ref_null_map[i] != new_null_map[i]); |
507 | 0 | } |
508 | 0 | if (is_changed) { |
509 | 0 | return Status::DataQualityError( |
510 | 0 | "null map is changed after calculation, input_column={}", |
511 | 0 | input_column->get_name()); |
512 | 0 | } |
513 | 0 | } |
514 | 0 | return Status::OK(); |
515 | 0 | } |
516 | | |
517 | | Status LinkedSchemaChange::process(RowsetReaderSharedPtr rowset_reader, RowsetWriter* rowset_writer, |
518 | | BaseTabletSPtr new_tablet, BaseTabletSPtr base_tablet, |
519 | | TabletSchemaSPtr base_tablet_schema, |
520 | 0 | TabletSchemaSPtr new_tablet_schema) { |
521 | 0 | Status status = rowset_writer->add_rowset_for_linked_schema_change(rowset_reader->rowset()); |
522 | 0 | if (!status) { |
523 | 0 | LOG(WARNING) << "fail to convert rowset." |
524 | 0 | << ", new_tablet=" << new_tablet->tablet_id() |
525 | 0 | << ", version=" << rowset_writer->version().first << "-" |
526 | 0 | << rowset_writer->version().second << ", error status " << status; |
527 | 0 | return status; |
528 | 0 | } |
529 | | // copy delete bitmap to new tablet. |
530 | 0 | if (new_tablet->keys_type() == UNIQUE_KEYS && new_tablet->enable_unique_key_merge_on_write()) { |
531 | 0 | DeleteBitmap origin_delete_bitmap(base_tablet->tablet_id()); |
532 | 0 | base_tablet->tablet_meta()->delete_bitmap().subset( |
533 | 0 | {rowset_reader->rowset()->rowset_id(), 0, 0}, |
534 | 0 | {rowset_reader->rowset()->rowset_id(), UINT32_MAX, INT64_MAX}, |
535 | 0 | &origin_delete_bitmap); |
536 | 0 | for (auto& iter : origin_delete_bitmap.delete_bitmap) { |
537 | 0 | int ret = new_tablet->tablet_meta()->delete_bitmap().set( |
538 | 0 | {rowset_writer->rowset_id(), std::get<1>(iter.first), std::get<2>(iter.first)}, |
539 | 0 | iter.second); |
540 | 0 | DCHECK(ret == 1); |
541 | 0 | } |
542 | 0 | } |
543 | 0 | return Status::OK(); |
544 | 0 | } |
545 | | |
546 | | Status VSchemaChangeDirectly::_inner_process(RowsetReaderSharedPtr rowset_reader, |
547 | | RowsetWriter* rowset_writer, BaseTabletSPtr new_tablet, |
548 | | TabletSchemaSPtr base_tablet_schema, |
549 | 0 | TabletSchemaSPtr new_tablet_schema) { |
550 | 0 | bool eof = false; |
551 | 0 | do { |
552 | 0 | auto new_block = vectorized::Block::create_unique(new_tablet_schema->create_block()); |
553 | 0 | auto ref_block = vectorized::Block::create_unique(base_tablet_schema->create_block()); |
554 | |
|
555 | 0 | auto st = rowset_reader->next_block(ref_block.get()); |
556 | 0 | if (!st) { |
557 | 0 | if (st.is<ErrorCode::END_OF_FILE>()) { |
558 | 0 | if (ref_block->rows() == 0) { |
559 | 0 | break; |
560 | 0 | } else { |
561 | 0 | eof = true; |
562 | 0 | } |
563 | 0 | } else { |
564 | 0 | return st; |
565 | 0 | } |
566 | 0 | } |
567 | | |
568 | 0 | RETURN_IF_ERROR(_changer.change_block(ref_block.get(), new_block.get())); |
569 | 0 | RETURN_IF_ERROR(rowset_writer->add_block(new_block.get())); |
570 | 0 | } while (!eof); |
571 | | |
572 | 0 | RETURN_IF_ERROR(rowset_writer->flush()); |
573 | 0 | return Status::OK(); |
574 | 0 | } |
575 | | |
576 | | VBaseSchemaChangeWithSorting::VBaseSchemaChangeWithSorting(const BlockChanger& changer, |
577 | | size_t memory_limitation) |
578 | | : _changer(changer), |
579 | | _memory_limitation(memory_limitation), |
580 | 0 | _temp_delta_versions(Version::mock()) { |
581 | 0 | _mem_tracker = std::make_unique<MemTracker>( |
582 | 0 | fmt::format("VSchemaChangeWithSorting:changer={}", std::to_string(int64(&changer)))); |
583 | 0 | } |
584 | | |
585 | | Status VBaseSchemaChangeWithSorting::_inner_process(RowsetReaderSharedPtr rowset_reader, |
586 | | RowsetWriter* rowset_writer, |
587 | | BaseTabletSPtr new_tablet, |
588 | | TabletSchemaSPtr base_tablet_schema, |
589 | 0 | TabletSchemaSPtr new_tablet_schema) { |
590 | | // for internal sorting |
591 | 0 | std::vector<std::unique_ptr<vectorized::Block>> blocks; |
592 | |
|
593 | 0 | RowsetSharedPtr rowset = rowset_reader->rowset(); |
594 | 0 | SegmentsOverlapPB segments_overlap = rowset->rowset_meta()->segments_overlap(); |
595 | 0 | int64_t newest_write_timestamp = rowset->newest_write_timestamp(); |
596 | 0 | _temp_delta_versions.first = _temp_delta_versions.second; |
597 | 0 | _src_rowsets.clear(); // init _src_rowsets |
598 | 0 | auto create_rowset = [&]() -> Status { |
599 | 0 | if (blocks.empty()) { |
600 | 0 | return Status::OK(); |
601 | 0 | } |
602 | | |
603 | 0 | auto rowset = DORIS_TRY(_internal_sorting( |
604 | 0 | blocks, Version(_temp_delta_versions.second, _temp_delta_versions.second), |
605 | 0 | newest_write_timestamp, new_tablet, BETA_ROWSET, segments_overlap, |
606 | 0 | new_tablet_schema)); |
607 | 0 | _src_rowsets.push_back(std::move(rowset)); |
608 | 0 | for (auto& block : blocks) { |
609 | 0 | _mem_tracker->release(block->allocated_bytes()); |
610 | 0 | } |
611 | 0 | blocks.clear(); |
612 | | |
613 | | // increase temp version |
614 | 0 | _temp_delta_versions.second++; |
615 | 0 | return Status::OK(); |
616 | 0 | }; |
617 | |
|
618 | 0 | auto new_block = vectorized::Block::create_unique(new_tablet_schema->create_block()); |
619 | |
|
620 | 0 | bool eof = false; |
621 | 0 | do { |
622 | 0 | auto ref_block = vectorized::Block::create_unique(base_tablet_schema->create_block()); |
623 | 0 | auto st = rowset_reader->next_block(ref_block.get()); |
624 | 0 | if (!st) { |
625 | 0 | if (st.is<ErrorCode::END_OF_FILE>()) { |
626 | 0 | if (ref_block->rows() == 0) { |
627 | 0 | break; |
628 | 0 | } else { |
629 | 0 | eof = true; |
630 | 0 | } |
631 | 0 | } else { |
632 | 0 | return st; |
633 | 0 | } |
634 | 0 | } |
635 | | |
636 | 0 | RETURN_IF_ERROR(_changer.change_block(ref_block.get(), new_block.get())); |
637 | | |
638 | 0 | constexpr double HOLD_BLOCK_MEMORY_RATE = |
639 | 0 | 0.66; // Reserve some memory for use by other parts of this job |
640 | 0 | if (_mem_tracker->consumption() + new_block->allocated_bytes() > _memory_limitation || |
641 | 0 | _mem_tracker->consumption() > _memory_limitation * HOLD_BLOCK_MEMORY_RATE || |
642 | 0 | DebugPoints::instance()->is_enable( |
643 | 0 | "VBaseSchemaChangeWithSorting._inner_process.create_rowset")) { |
644 | 0 | RETURN_IF_ERROR(create_rowset()); |
645 | | |
646 | 0 | if (_mem_tracker->consumption() + new_block->allocated_bytes() > _memory_limitation) { |
647 | 0 | return Status::Error<INVALID_ARGUMENT>( |
648 | 0 | "Memory limitation is too small for Schema Change. _memory_limitation={}, " |
649 | 0 | "new_block->allocated_bytes()={}, consumption={}", |
650 | 0 | _memory_limitation, new_block->allocated_bytes(), |
651 | 0 | _mem_tracker->consumption()); |
652 | 0 | } |
653 | 0 | } |
654 | 0 | _mem_tracker->consume(new_block->allocated_bytes()); |
655 | | |
656 | | // move unique ptr |
657 | 0 | blocks.push_back(vectorized::Block::create_unique(new_tablet_schema->create_block())); |
658 | 0 | swap(blocks.back(), new_block); |
659 | 0 | } while (!eof); |
660 | | |
661 | 0 | RETURN_IF_ERROR(create_rowset()); |
662 | | |
663 | 0 | if (_src_rowsets.empty()) { |
664 | 0 | RETURN_IF_ERROR(rowset_writer->flush()); |
665 | 0 | } else { |
666 | 0 | RETURN_IF_ERROR( |
667 | 0 | _external_sorting(_src_rowsets, rowset_writer, new_tablet, new_tablet_schema)); |
668 | 0 | } |
669 | | |
670 | 0 | return Status::OK(); |
671 | 0 | } |
672 | | |
673 | | Result<RowsetSharedPtr> VBaseSchemaChangeWithSorting::_internal_sorting( |
674 | | const std::vector<std::unique_ptr<vectorized::Block>>& blocks, const Version& version, |
675 | | int64_t newest_write_timestamp, BaseTabletSPtr new_tablet, RowsetTypePB new_rowset_type, |
676 | 0 | SegmentsOverlapPB segments_overlap, TabletSchemaSPtr new_tablet_schema) { |
677 | 0 | uint64_t merged_rows = 0; |
678 | 0 | MultiBlockMerger merger(new_tablet); |
679 | 0 | RowsetWriterContext context; |
680 | 0 | context.version = version; |
681 | 0 | context.rowset_state = VISIBLE; |
682 | 0 | context.segments_overlap = segments_overlap; |
683 | 0 | context.tablet_schema = new_tablet_schema; |
684 | 0 | context.newest_write_timestamp = newest_write_timestamp; |
685 | 0 | context.write_type = DataWriteType::TYPE_SCHEMA_CHANGE; |
686 | 0 | std::unique_ptr<RowsetWriter> rowset_writer; |
687 | | // TODO(plat1ko): Use monad op |
688 | 0 | if (auto result = new_tablet->create_rowset_writer(context, false); !result.has_value()) |
689 | 0 | [[unlikely]] { |
690 | 0 | return unexpected(std::move(result).error()); |
691 | 0 | } else { |
692 | 0 | rowset_writer = std::move(result).value(); |
693 | 0 | } |
694 | 0 | RETURN_IF_ERROR_RESULT(merger.merge(blocks, rowset_writer.get(), &merged_rows)); |
695 | 0 | _add_merged_rows(merged_rows); |
696 | 0 | RowsetSharedPtr rowset; |
697 | 0 | RETURN_IF_ERROR_RESULT(rowset_writer->build(rowset)); |
698 | 0 | return rowset; |
699 | 0 | } |
700 | | |
701 | | Result<RowsetSharedPtr> VLocalSchemaChangeWithSorting::_internal_sorting( |
702 | | const std::vector<std::unique_ptr<vectorized::Block>>& blocks, const Version& version, |
703 | | int64_t newest_write_timestamp, BaseTabletSPtr new_tablet, RowsetTypePB new_rowset_type, |
704 | 0 | SegmentsOverlapPB segments_overlap, TabletSchemaSPtr new_tablet_schema) { |
705 | 0 | uint64_t merged_rows = 0; |
706 | 0 | MultiBlockMerger merger(new_tablet); |
707 | 0 | RowsetWriterContext context; |
708 | 0 | context.version = version; |
709 | 0 | context.rowset_state = VISIBLE; |
710 | 0 | context.segments_overlap = segments_overlap; |
711 | 0 | context.tablet_schema = new_tablet_schema; |
712 | 0 | context.newest_write_timestamp = newest_write_timestamp; |
713 | 0 | context.write_type = DataWriteType::TYPE_SCHEMA_CHANGE; |
714 | 0 | std::unique_ptr<RowsetWriter> rowset_writer; |
715 | | // TODO(plat1ko): Use monad op |
716 | 0 | if (auto result = new_tablet->create_rowset_writer(context, false); !result.has_value()) |
717 | 0 | [[unlikely]] { |
718 | 0 | return unexpected(std::move(result).error()); |
719 | 0 | } else { |
720 | 0 | rowset_writer = std::move(result).value(); |
721 | 0 | } |
722 | 0 | auto guard = _local_storage_engine.pending_local_rowsets().add(context.rowset_id); |
723 | 0 | _pending_rs_guards.push_back(std::move(guard)); |
724 | 0 | RETURN_IF_ERROR_RESULT(merger.merge(blocks, rowset_writer.get(), &merged_rows)); |
725 | 0 | _add_merged_rows(merged_rows); |
726 | 0 | RowsetSharedPtr rowset; |
727 | 0 | RETURN_IF_ERROR_RESULT(rowset_writer->build(rowset)); |
728 | 0 | return rowset; |
729 | 0 | } |
730 | | |
731 | | Status VBaseSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_rowsets, |
732 | | RowsetWriter* rowset_writer, |
733 | | BaseTabletSPtr new_tablet, |
734 | 0 | TabletSchemaSPtr new_tablet_schema) { |
735 | 0 | std::vector<RowsetReaderSharedPtr> rs_readers; |
736 | 0 | for (auto& rowset : src_rowsets) { |
737 | 0 | RowsetReaderSharedPtr rs_reader; |
738 | 0 | RETURN_IF_ERROR(rowset->create_reader(&rs_reader)); |
739 | 0 | rs_readers.push_back(rs_reader); |
740 | 0 | } |
741 | | |
742 | 0 | Merger::Statistics stats; |
743 | 0 | if (!new_tablet_schema->cluster_key_uids().empty()) { |
744 | | // schema change read rowsets with delete bitmap, so there should be no duplicated keys |
745 | | // RETURN_IF_ERROR(Compaction::update_delete_bitmap()); |
746 | 0 | int64_t way_num = 0; |
747 | 0 | int64_t input_rowsets_data_size = 0; |
748 | 0 | int64_t input_row_num = 0; |
749 | 0 | for (auto& rowset : src_rowsets) { |
750 | 0 | way_num += rowset->rowset_meta()->get_merge_way_num(); |
751 | 0 | input_rowsets_data_size += rowset->data_disk_size(); |
752 | 0 | input_row_num += rowset->num_rows(); |
753 | 0 | } |
754 | 0 | int64_t avg_segment_rows = config::vertical_compaction_max_segment_size / |
755 | 0 | (input_rowsets_data_size / (input_row_num + 1) + 1); |
756 | 0 | RETURN_IF_ERROR(Merger::vertical_merge_rowsets( |
757 | 0 | new_tablet, ReaderType::READER_ALTER_TABLE, *new_tablet_schema, rs_readers, |
758 | 0 | rowset_writer, avg_segment_rows, way_num, &stats)); |
759 | 0 | } else { |
760 | 0 | RETURN_IF_ERROR(Merger::vmerge_rowsets(new_tablet, ReaderType::READER_ALTER_TABLE, |
761 | 0 | *new_tablet_schema, rs_readers, rowset_writer, |
762 | 0 | &stats)); |
763 | 0 | } |
764 | 0 | _add_merged_rows(stats.merged_rows); |
765 | 0 | _add_filtered_rows(stats.filtered_rows); |
766 | 0 | return Status::OK(); |
767 | 0 | } |
768 | | |
769 | | Status VLocalSchemaChangeWithSorting::_inner_process(RowsetReaderSharedPtr rowset_reader, |
770 | | RowsetWriter* rowset_writer, |
771 | | BaseTabletSPtr new_tablet, |
772 | | TabletSchemaSPtr base_tablet_schema, |
773 | 0 | TabletSchemaSPtr new_tablet_schema) { |
774 | 0 | Defer defer {[&]() { |
775 | | // remove the intermediate rowsets generated by internal sorting |
776 | 0 | for (auto& row_set : _src_rowsets) { |
777 | 0 | _local_storage_engine.add_unused_rowset(row_set); |
778 | 0 | } |
779 | 0 | }}; |
780 | 0 | _pending_rs_guards.clear(); |
781 | 0 | return VBaseSchemaChangeWithSorting::_inner_process(rowset_reader, rowset_writer, new_tablet, |
782 | 0 | base_tablet_schema, new_tablet_schema); |
783 | 0 | } |
784 | | |
785 | 0 | Status SchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { |
786 | 0 | if (!request.__isset.desc_tbl) { |
787 | 0 | return Status::Error<INVALID_ARGUMENT>( |
788 | 0 | "desc_tbl is not set. Maybe the FE version is not equal to the BE " |
789 | 0 | "version."); |
790 | 0 | } |
791 | 0 | if (_base_tablet == nullptr) { |
792 | 0 | return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", |
793 | 0 | request.base_tablet_id); |
794 | 0 | } |
795 | 0 | if (_new_tablet == nullptr) { |
796 | 0 | return Status::Error<TABLE_NOT_FOUND>("fail to find new tablet. new_tablet={}", |
797 | 0 | request.new_tablet_id); |
798 | 0 | } |
799 | | |
800 | 0 | LOG(INFO) << "begin to do request alter tablet: base_tablet_id=" << request.base_tablet_id |
801 | 0 | << ", new_tablet_id=" << request.new_tablet_id |
802 | 0 | << ", alter_version=" << request.alter_version; |
803 | | |
804 | | // Lock schema_change_lock util schema change info is stored in tablet header |
805 | 0 | static constexpr long TRY_LOCK_TIMEOUT = 30; |
806 | 0 | std::unique_lock schema_change_lock(_base_tablet->get_schema_change_lock(), std::defer_lock); |
807 | 0 | bool owns_lock = schema_change_lock.try_lock_for(std::chrono::seconds(TRY_LOCK_TIMEOUT)); |
808 | |
|
809 | 0 | if (!owns_lock) { |
810 | 0 | return Status::Error<TRY_LOCK_FAILED>( |
811 | 0 | "Failed to obtain schema change lock, there might be inverted index being " |
812 | 0 | "built or cooldown runnning on base_tablet={}", |
813 | 0 | request.base_tablet_id); |
814 | 0 | } |
815 | | |
816 | 0 | Status res = _do_process_alter_tablet(request); |
817 | 0 | LOG(INFO) << "finished alter tablet process, res=" << res; |
818 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob::process_alter_tablet.leave.sleep", { sleep(5); }); |
819 | 0 | return res; |
820 | 0 | } |
821 | | |
822 | | SchemaChangeJob::SchemaChangeJob(StorageEngine& local_storage_engine, |
823 | | const TAlterTabletReqV2& request, const std::string& job_id) |
824 | 0 | : _local_storage_engine(local_storage_engine) { |
825 | 0 | _base_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.base_tablet_id); |
826 | 0 | _new_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.new_tablet_id); |
827 | 0 | if (_base_tablet && _new_tablet) { |
828 | 0 | _base_tablet_schema = std::make_shared<TabletSchema>(); |
829 | 0 | _base_tablet_schema->update_tablet_columns(*_base_tablet->tablet_schema(), request.columns); |
830 | | // The request only include column info, do not include bitmap or bloomfilter index info, |
831 | | // So we also need to copy index info from the real base tablet |
832 | 0 | _base_tablet_schema->update_index_info_from(*_base_tablet->tablet_schema()); |
833 | | // During a schema change, the extracted columns of a variant should not be included in the tablet schema. |
834 | | // This is because the schema change for a variant needs to ignore the extracted columns. |
835 | | // Otherwise, the schema types in different rowsets might be inconsistent. When performing a schema change, |
836 | | // the complete variant is constructed by reading all the sub-columns of the variant. |
837 | 0 | _new_tablet_schema = _new_tablet->tablet_schema()->copy_without_variant_extracted_columns(); |
838 | 0 | } |
839 | 0 | _job_id = job_id; |
840 | 0 | } |
841 | | |
842 | | // In the past schema change and rollup will create new tablet and will wait for txns starting before the task to finished |
843 | | // It will cost a lot of time to wait and the task is very difficult to understand. |
844 | | // In alter task v2, FE will call BE to create tablet and send an alter task to BE to convert historical data. |
845 | | // The admin should upgrade all BE and then upgrade FE. |
846 | | // Should delete the old code after upgrade finished. |
847 | 0 | Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { |
848 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob._do_process_alter_tablet.sleep", { sleep(10); }) |
849 | 0 | Status res; |
850 | 0 | signal::tablet_id = _base_tablet->get_table_id(); |
851 | | |
852 | | // check if tablet's state is not_ready, if it is ready, it means the tablet already finished |
853 | | // check whether the tablet's max continuous version == request.version |
854 | 0 | if (_new_tablet->tablet_state() != TABLET_NOTREADY) { |
855 | 0 | res = _validate_alter_result(request); |
856 | 0 | LOG(INFO) << "tablet's state=" << _new_tablet->tablet_state() |
857 | 0 | << " the convert job already finished, check its version" |
858 | 0 | << " res=" << res; |
859 | 0 | return res; |
860 | 0 | } |
861 | 0 | _new_tablet->set_alter_failed(false); |
862 | 0 | Defer defer([this] { |
863 | | // if tablet state is not TABLET_RUNNING when return, indicates that alter has failed. |
864 | 0 | if (_new_tablet->tablet_state() != TABLET_RUNNING) { |
865 | 0 | _new_tablet->set_alter_failed(true); |
866 | 0 | } |
867 | 0 | }); |
868 | |
|
869 | 0 | LOG(INFO) << "finish to validate alter tablet request. begin to convert data from base tablet " |
870 | 0 | "to new tablet" |
871 | 0 | << " base_tablet=" << _base_tablet->tablet_id() |
872 | 0 | << " new_tablet=" << _new_tablet->tablet_id(); |
873 | |
|
874 | 0 | std::shared_lock base_migration_rlock(_base_tablet->get_migration_lock(), std::try_to_lock); |
875 | 0 | if (!base_migration_rlock.owns_lock()) { |
876 | 0 | return Status::Error<TRY_LOCK_FAILED>( |
877 | 0 | "SchemaChangeJob::_do_process_alter_tablet get lock failed"); |
878 | 0 | } |
879 | 0 | std::shared_lock new_migration_rlock(_new_tablet->get_migration_lock(), std::try_to_lock); |
880 | 0 | if (!new_migration_rlock.owns_lock()) { |
881 | 0 | return Status::Error<TRY_LOCK_FAILED>( |
882 | 0 | "SchemaChangeJob::_do_process_alter_tablet get lock failed"); |
883 | 0 | } |
884 | | |
885 | 0 | std::vector<Version> versions_to_be_changed; |
886 | 0 | int64_t end_version = -1; |
887 | | // reader_context is stack variables, it's lifetime should keep the same |
888 | | // with rs_readers |
889 | 0 | RowsetReaderContext reader_context; |
890 | 0 | std::vector<RowSetSplits> rs_splits; |
891 | | // delete handlers for new tablet |
892 | 0 | DeleteHandler delete_handler; |
893 | 0 | std::vector<ColumnId> return_columns; |
894 | | |
895 | | // Use tablet schema directly from base tablet, they are the newest schema, not contain |
896 | | // dropped column during light weight schema change. |
897 | | // But the tablet schema in base tablet maybe not the latest from FE, so that if fe pass through |
898 | | // a tablet schema, then use request schema. |
899 | 0 | size_t num_cols = |
900 | 0 | request.columns.empty() ? _base_tablet_schema->num_columns() : request.columns.size(); |
901 | 0 | return_columns.resize(num_cols); |
902 | 0 | for (int i = 0; i < num_cols; ++i) { |
903 | 0 | return_columns[i] = i; |
904 | 0 | } |
905 | 0 | std::vector<uint32_t> cluster_key_idxes; |
906 | |
|
907 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob::_do_process_alter_tablet.block", DBUG_BLOCK); |
908 | | |
909 | | // begin to find deltas to convert from base tablet to new tablet so that |
910 | | // obtain base tablet and new tablet's push lock and header write lock to prevent loading data |
911 | 0 | { |
912 | 0 | std::lock_guard base_tablet_lock(_base_tablet->get_push_lock()); |
913 | 0 | std::lock_guard new_tablet_lock(_new_tablet->get_push_lock()); |
914 | 0 | std::lock_guard base_tablet_wlock(_base_tablet->get_header_lock()); |
915 | 0 | SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
916 | 0 | std::lock_guard<std::shared_mutex> new_tablet_wlock(_new_tablet->get_header_lock()); |
917 | |
|
918 | 0 | do { |
919 | 0 | RowsetSharedPtr max_rowset; |
920 | | // get history data to be converted and it will check if there is hold in base tablet |
921 | 0 | res = _get_versions_to_be_changed(&versions_to_be_changed, &max_rowset); |
922 | 0 | if (!res) { |
923 | 0 | LOG(WARNING) << "fail to get version to be changed. res=" << res; |
924 | 0 | break; |
925 | 0 | } |
926 | | |
927 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob.process_alter_tablet.alter_fail", { |
928 | 0 | res = Status::InternalError( |
929 | 0 | "inject alter tablet failed. base_tablet={}, new_tablet={}", |
930 | 0 | request.base_tablet_id, request.new_tablet_id); |
931 | 0 | LOG(WARNING) << "inject error. res=" << res; |
932 | 0 | break; |
933 | 0 | }); |
934 | | |
935 | | // should check the max_version >= request.alter_version, if not the convert is useless |
936 | 0 | if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) { |
937 | 0 | res = Status::InternalError( |
938 | 0 | "base tablet's max version={} is less than request version={}", |
939 | 0 | (max_rowset == nullptr ? 0 : max_rowset->end_version()), |
940 | 0 | request.alter_version); |
941 | 0 | break; |
942 | 0 | } |
943 | | // before calculating version_to_be_changed, |
944 | | // remove all data from new tablet, prevent to rewrite data(those double pushed when wait) |
945 | 0 | LOG(INFO) << "begin to remove all data before end version from new tablet to prevent " |
946 | 0 | "rewrite." |
947 | 0 | << " new_tablet=" << _new_tablet->tablet_id() |
948 | 0 | << ", end_version=" << max_rowset->end_version(); |
949 | 0 | std::vector<RowsetSharedPtr> rowsets_to_delete; |
950 | 0 | std::vector<std::pair<Version, RowsetSharedPtr>> version_rowsets; |
951 | 0 | _new_tablet->acquire_version_and_rowsets(&version_rowsets); |
952 | 0 | std::sort(version_rowsets.begin(), version_rowsets.end(), |
953 | 0 | [](const std::pair<Version, RowsetSharedPtr>& l, |
954 | 0 | const std::pair<Version, RowsetSharedPtr>& r) { |
955 | 0 | return l.first.first < r.first.first; |
956 | 0 | }); |
957 | 0 | for (auto& pair : version_rowsets) { |
958 | 0 | if (pair.first.second <= max_rowset->end_version()) { |
959 | 0 | rowsets_to_delete.push_back(pair.second); |
960 | 0 | } else if (pair.first.first <= max_rowset->end_version()) { |
961 | | // If max version is [X-10] and new tablet has version [7-9][10-12], |
962 | | // we only can remove [7-9] from new tablet. If we add [X-10] to new tablet, it will has version |
963 | | // cross: [X-10] [10-12]. |
964 | | // So, we should return OLAP_ERR_VERSION_ALREADY_MERGED for fast fail. |
965 | 0 | return Status::Error<VERSION_ALREADY_MERGED>( |
966 | 0 | "New tablet has a version {} crossing base tablet's max_version={}", |
967 | 0 | pair.first.to_string(), max_rowset->end_version()); |
968 | 0 | } |
969 | 0 | } |
970 | 0 | std::vector<RowsetSharedPtr> empty_vec; |
971 | 0 | RETURN_IF_ERROR(_new_tablet->delete_rowsets(rowsets_to_delete, false)); |
972 | | // inherit cumulative_layer_point from base_tablet |
973 | | // check if new_tablet.ce_point > base_tablet.ce_point? |
974 | 0 | _new_tablet->set_cumulative_layer_point(-1); |
975 | | // save tablet meta |
976 | 0 | _new_tablet->save_meta(); |
977 | 0 | for (auto& rowset : rowsets_to_delete) { |
978 | | // do not call rowset.remove directly, using gc thread to delete it |
979 | 0 | _local_storage_engine.add_unused_rowset(rowset); |
980 | 0 | } |
981 | | |
982 | | // init one delete handler |
983 | 0 | for (auto& version : versions_to_be_changed) { |
984 | 0 | end_version = std::max(end_version, version.second); |
985 | 0 | } |
986 | | |
987 | | // acquire data sources correspond to history versions |
988 | 0 | RETURN_IF_ERROR( |
989 | 0 | _base_tablet->capture_rs_readers_unlocked(versions_to_be_changed, &rs_splits)); |
990 | 0 | if (rs_splits.empty()) { |
991 | 0 | res = Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>( |
992 | 0 | "fail to acquire all data sources. version_num={}, data_source_num={}", |
993 | 0 | versions_to_be_changed.size(), rs_splits.size()); |
994 | 0 | break; |
995 | 0 | } |
996 | 0 | std::vector<RowsetMetaSharedPtr> del_preds; |
997 | 0 | for (auto&& split : rs_splits) { |
998 | 0 | const auto& rs_meta = split.rs_reader->rowset()->rowset_meta(); |
999 | 0 | if (!rs_meta->has_delete_predicate() || rs_meta->start_version() > end_version) { |
1000 | 0 | continue; |
1001 | 0 | } |
1002 | 0 | _base_tablet_schema->merge_dropped_columns(*rs_meta->tablet_schema()); |
1003 | 0 | del_preds.push_back(rs_meta); |
1004 | 0 | } |
1005 | 0 | res = delete_handler.init(_base_tablet_schema, del_preds, end_version); |
1006 | 0 | if (!res) { |
1007 | 0 | LOG(WARNING) << "init delete handler failed. base_tablet=" |
1008 | 0 | << _base_tablet->tablet_id() << ", end_version=" << end_version; |
1009 | 0 | break; |
1010 | 0 | } |
1011 | | |
1012 | 0 | reader_context.reader_type = ReaderType::READER_ALTER_TABLE; |
1013 | 0 | reader_context.tablet_schema = _base_tablet_schema; |
1014 | 0 | reader_context.need_ordered_result = true; |
1015 | 0 | reader_context.delete_handler = &delete_handler; |
1016 | 0 | reader_context.return_columns = &return_columns; |
1017 | 0 | reader_context.sequence_id_idx = reader_context.tablet_schema->sequence_col_idx(); |
1018 | 0 | reader_context.is_unique = _base_tablet->keys_type() == UNIQUE_KEYS; |
1019 | 0 | reader_context.batch_size = ALTER_TABLE_BATCH_SIZE; |
1020 | 0 | reader_context.delete_bitmap = &_base_tablet->tablet_meta()->delete_bitmap(); |
1021 | 0 | reader_context.version = Version(0, end_version); |
1022 | 0 | if (!_base_tablet_schema->cluster_key_uids().empty()) { |
1023 | 0 | for (const auto& uid : _base_tablet_schema->cluster_key_uids()) { |
1024 | 0 | cluster_key_idxes.emplace_back(_base_tablet_schema->field_index(uid)); |
1025 | 0 | } |
1026 | 0 | reader_context.read_orderby_key_columns = &cluster_key_idxes; |
1027 | 0 | reader_context.is_unique = false; |
1028 | 0 | reader_context.sequence_id_idx = -1; |
1029 | 0 | } |
1030 | 0 | for (auto& rs_split : rs_splits) { |
1031 | 0 | res = rs_split.rs_reader->init(&reader_context); |
1032 | 0 | if (!res) { |
1033 | 0 | LOG(WARNING) << "failed to init rowset reader: " << _base_tablet->tablet_id(); |
1034 | 0 | break; |
1035 | 0 | } |
1036 | 0 | } |
1037 | 0 | } while (false); |
1038 | 0 | } |
1039 | | |
1040 | 0 | do { |
1041 | 0 | if (!res) { |
1042 | 0 | break; |
1043 | 0 | } |
1044 | 0 | SchemaChangeParams sc_params; |
1045 | |
|
1046 | 0 | RETURN_IF_ERROR( |
1047 | 0 | DescriptorTbl::create(&sc_params.pool, request.desc_tbl, &sc_params.desc_tbl)); |
1048 | 0 | sc_params.ref_rowset_readers.reserve(rs_splits.size()); |
1049 | 0 | for (RowSetSplits& split : rs_splits) { |
1050 | 0 | sc_params.ref_rowset_readers.emplace_back(split.rs_reader); |
1051 | 0 | } |
1052 | 0 | sc_params.delete_handler = &delete_handler; |
1053 | 0 | sc_params.be_exec_version = request.be_exec_version; |
1054 | 0 | DCHECK(request.__isset.alter_tablet_type); |
1055 | 0 | switch (request.alter_tablet_type) { |
1056 | 0 | case TAlterTabletType::SCHEMA_CHANGE: |
1057 | 0 | sc_params.alter_tablet_type = AlterTabletType::SCHEMA_CHANGE; |
1058 | 0 | break; |
1059 | 0 | case TAlterTabletType::ROLLUP: |
1060 | 0 | sc_params.alter_tablet_type = AlterTabletType::ROLLUP; |
1061 | 0 | break; |
1062 | 0 | case TAlterTabletType::MIGRATION: |
1063 | 0 | sc_params.alter_tablet_type = AlterTabletType::MIGRATION; |
1064 | 0 | break; |
1065 | 0 | } |
1066 | 0 | if (request.__isset.materialized_view_params) { |
1067 | 0 | for (auto item : request.materialized_view_params) { |
1068 | 0 | AlterMaterializedViewParam mv_param; |
1069 | 0 | mv_param.column_name = item.column_name; |
1070 | |
|
1071 | 0 | if (item.__isset.mv_expr) { |
1072 | 0 | mv_param.expr = std::make_shared<TExpr>(item.mv_expr); |
1073 | 0 | } |
1074 | 0 | sc_params.materialized_params_map.insert( |
1075 | 0 | std::make_pair(to_lower(item.column_name), mv_param)); |
1076 | 0 | } |
1077 | 0 | } |
1078 | 0 | { |
1079 | 0 | std::lock_guard<std::shared_mutex> wrlock(_mutex); |
1080 | 0 | _tablet_ids_in_converting.insert(_new_tablet->tablet_id()); |
1081 | 0 | } |
1082 | 0 | int64_t real_alter_version = 0; |
1083 | 0 | sc_params.enable_unique_key_merge_on_write = |
1084 | 0 | _new_tablet->enable_unique_key_merge_on_write(); |
1085 | 0 | res = _convert_historical_rowsets(sc_params, &real_alter_version); |
1086 | 0 | { |
1087 | 0 | std::lock_guard<std::shared_mutex> wrlock(_mutex); |
1088 | 0 | _tablet_ids_in_converting.erase(_new_tablet->tablet_id()); |
1089 | 0 | } |
1090 | 0 | if (!res) { |
1091 | 0 | break; |
1092 | 0 | } |
1093 | | |
1094 | 0 | DCHECK_GE(real_alter_version, request.alter_version); |
1095 | |
|
1096 | 0 | if (_new_tablet->keys_type() == UNIQUE_KEYS && |
1097 | 0 | _new_tablet->enable_unique_key_merge_on_write()) { |
1098 | 0 | res = _calc_delete_bitmap_for_mow_table(real_alter_version); |
1099 | 0 | if (!res) { |
1100 | 0 | break; |
1101 | 0 | } |
1102 | 0 | } else { |
1103 | | // set state to ready |
1104 | 0 | std::lock_guard<std::shared_mutex> new_wlock(_new_tablet->get_header_lock()); |
1105 | 0 | SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
1106 | 0 | res = _new_tablet->set_tablet_state(TabletState::TABLET_RUNNING); |
1107 | 0 | if (!res) { |
1108 | 0 | break; |
1109 | 0 | } |
1110 | 0 | _new_tablet->save_meta(); |
1111 | 0 | } |
1112 | 0 | } while (false); |
1113 | | |
1114 | 0 | if (res) { |
1115 | | // _validate_alter_result should be outside the above while loop. |
1116 | | // to avoid requiring the header lock twice. |
1117 | 0 | res = _validate_alter_result(request); |
1118 | 0 | } |
1119 | | |
1120 | | // if failed convert history data, then just remove the new tablet |
1121 | 0 | if (!res) { |
1122 | 0 | LOG(WARNING) << "failed to alter tablet. base_tablet=" << _base_tablet->tablet_id() |
1123 | 0 | << ", drop new_tablet=" << _new_tablet->tablet_id(); |
1124 | | // do not drop the new tablet and its data. GC thread will |
1125 | 0 | } |
1126 | |
|
1127 | 0 | return res; |
1128 | 0 | } |
1129 | | |
1130 | 0 | bool SchemaChangeJob::tablet_in_converting(int64_t tablet_id) { |
1131 | 0 | std::shared_lock rdlock(_mutex); |
1132 | 0 | return _tablet_ids_in_converting.find(tablet_id) != _tablet_ids_in_converting.end(); |
1133 | 0 | } |
1134 | | |
1135 | | Status SchemaChangeJob::_get_versions_to_be_changed(std::vector<Version>* versions_to_be_changed, |
1136 | 0 | RowsetSharedPtr* max_rowset) { |
1137 | 0 | RowsetSharedPtr rowset = _base_tablet->get_rowset_with_max_version(); |
1138 | 0 | if (rowset == nullptr) { |
1139 | 0 | return Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>("Tablet has no version. base_tablet={}", |
1140 | 0 | _base_tablet->tablet_id()); |
1141 | 0 | } |
1142 | 0 | *max_rowset = rowset; |
1143 | |
|
1144 | 0 | RETURN_IF_ERROR(_base_tablet->capture_consistent_versions_unlocked( |
1145 | 0 | Version(0, rowset->version().second), versions_to_be_changed, false, false)); |
1146 | | |
1147 | 0 | return Status::OK(); |
1148 | 0 | } |
1149 | | |
1150 | | // The `real_alter_version` parameter indicates that the version of [0-real_alter_version] is |
1151 | | // converted from a base tablet, only used for the mow table now. |
1152 | | Status SchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params, |
1153 | 0 | int64_t* real_alter_version) { |
1154 | 0 | LOG(INFO) << "begin to convert historical rowsets for new_tablet from base_tablet." |
1155 | 0 | << " base_tablet=" << _base_tablet->tablet_id() |
1156 | 0 | << ", new_tablet=" << _new_tablet->tablet_id() << ", job_id=" << _job_id; |
1157 | | |
1158 | | // find end version |
1159 | 0 | int32_t end_version = -1; |
1160 | 0 | for (const auto& ref_rowset_reader : sc_params.ref_rowset_readers) { |
1161 | 0 | if (ref_rowset_reader->version().second > end_version) { |
1162 | 0 | end_version = ref_rowset_reader->version().second; |
1163 | 0 | } |
1164 | 0 | } |
1165 | | |
1166 | | // Add filter information in change, and filter column information will be set in parse_request |
1167 | | // And filter some data every time the row block changes |
1168 | 0 | BlockChanger changer(_new_tablet_schema, *sc_params.desc_tbl); |
1169 | |
|
1170 | 0 | bool sc_sorting = false; |
1171 | 0 | bool sc_directly = false; |
1172 | | |
1173 | | // a.Parse the Alter request and convert it into an internal representation |
1174 | 0 | Status res = parse_request(sc_params, _base_tablet_schema.get(), _new_tablet_schema.get(), |
1175 | 0 | &changer, &sc_sorting, &sc_directly); |
1176 | 0 | LOG(INFO) << "schema change type, sc_sorting: " << sc_sorting |
1177 | 0 | << ", sc_directly: " << sc_directly << ", base_tablet=" << _base_tablet->tablet_id() |
1178 | 0 | << ", new_tablet=" << _new_tablet->tablet_id(); |
1179 | |
|
1180 | 0 | auto process_alter_exit = [&]() -> Status { |
1181 | 0 | { |
1182 | | // save tablet meta here because rowset meta is not saved during add rowset |
1183 | 0 | std::lock_guard new_wlock(_new_tablet->get_header_lock()); |
1184 | 0 | SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
1185 | 0 | _new_tablet->save_meta(); |
1186 | 0 | } |
1187 | 0 | if (res) { |
1188 | 0 | Version test_version(0, end_version); |
1189 | 0 | res = _new_tablet->check_version_integrity(test_version); |
1190 | 0 | } |
1191 | |
|
1192 | 0 | LOG(INFO) << "finish converting rowsets for new_tablet from base_tablet. " |
1193 | 0 | << "base_tablet=" << _base_tablet->tablet_id() |
1194 | 0 | << ", new_tablet=" << _new_tablet->tablet_id(); |
1195 | 0 | return res; |
1196 | 0 | }; |
1197 | |
|
1198 | 0 | if (!res) { |
1199 | 0 | LOG(WARNING) << "failed to parse the request. res=" << res; |
1200 | 0 | return process_alter_exit(); |
1201 | 0 | } |
1202 | | |
1203 | 0 | if (!sc_sorting && !sc_directly && sc_params.alter_tablet_type == AlterTabletType::ROLLUP) { |
1204 | 0 | res = Status::Error<SCHEMA_SCHEMA_INVALID>( |
1205 | 0 | "Don't support to add materialized view by linked schema change"); |
1206 | 0 | return process_alter_exit(); |
1207 | 0 | } |
1208 | | |
1209 | | // b. Generate historical data converter |
1210 | 0 | auto sc_procedure = _get_sc_procedure( |
1211 | 0 | changer, sc_sorting, sc_directly, |
1212 | 0 | _local_storage_engine.memory_limitation_bytes_per_thread_for_schema_change()); |
1213 | |
|
1214 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob::_convert_historical_rowsets.block", DBUG_BLOCK); |
1215 | | |
1216 | | // c.Convert historical data |
1217 | 0 | bool have_failure_rowset = false; |
1218 | 0 | for (const auto& rs_reader : sc_params.ref_rowset_readers) { |
1219 | | // set status for monitor |
1220 | | // As long as there is a new_table as running, ref table is set as running |
1221 | | // NOTE If the first sub_table fails first, it will continue to go as normal here |
1222 | | // When tablet create new rowset writer, it may change rowset type, in this case |
1223 | | // linked schema change will not be used. |
1224 | 0 | RowsetWriterContext context; |
1225 | 0 | context.version = rs_reader->version(); |
1226 | 0 | context.rowset_state = VISIBLE; |
1227 | 0 | context.segments_overlap = rs_reader->rowset()->rowset_meta()->segments_overlap(); |
1228 | 0 | context.tablet_schema = _new_tablet_schema; |
1229 | 0 | context.newest_write_timestamp = rs_reader->newest_write_timestamp(); |
1230 | |
|
1231 | 0 | if (!rs_reader->rowset()->is_local()) { |
1232 | 0 | context.storage_resource = |
1233 | 0 | *DORIS_TRY(rs_reader->rowset()->rowset_meta()->remote_storage_resource()); |
1234 | 0 | } |
1235 | | |
1236 | 0 | context.write_type = DataWriteType::TYPE_SCHEMA_CHANGE; |
1237 | | // TODO if support VerticalSegmentWriter, also need to handle cluster key primary key index |
1238 | 0 | bool vertical = false; |
1239 | 0 | if (sc_sorting && !_new_tablet->tablet_schema()->cluster_key_uids().empty()) { |
1240 | | // see VBaseSchemaChangeWithSorting::_external_sorting |
1241 | 0 | vertical = true; |
1242 | 0 | } |
1243 | 0 | auto result = _new_tablet->create_rowset_writer(context, vertical); |
1244 | 0 | if (!result.has_value()) { |
1245 | 0 | res = Status::Error<ROWSET_BUILDER_INIT>("create_rowset_writer failed, reason={}", |
1246 | 0 | result.error().to_string()); |
1247 | 0 | return process_alter_exit(); |
1248 | 0 | } |
1249 | 0 | auto rowset_writer = std::move(result).value(); |
1250 | 0 | auto pending_rs_guard = _local_storage_engine.add_pending_rowset(context); |
1251 | |
|
1252 | 0 | if (res = sc_procedure->process(rs_reader, rowset_writer.get(), _new_tablet, _base_tablet, |
1253 | 0 | _base_tablet_schema, _new_tablet_schema); |
1254 | 0 | !res) { |
1255 | 0 | LOG(WARNING) << "failed to process the version." |
1256 | 0 | << " version=" << rs_reader->version().first << "-" |
1257 | 0 | << rs_reader->version().second << ", " << res.to_string(); |
1258 | 0 | return process_alter_exit(); |
1259 | 0 | } |
1260 | | // Add the new version of the data to the header |
1261 | | // In order to prevent the occurrence of deadlock, we must first lock the old table, and then lock the new table |
1262 | 0 | std::lock_guard lock(_new_tablet->get_push_lock()); |
1263 | 0 | RowsetSharedPtr new_rowset; |
1264 | 0 | if (!(res = rowset_writer->build(new_rowset)).ok()) { |
1265 | 0 | LOG(WARNING) << "failed to build rowset, exit alter process"; |
1266 | 0 | return process_alter_exit(); |
1267 | 0 | } |
1268 | 0 | res = _new_tablet->add_rowset(new_rowset); |
1269 | 0 | if (res.is<PUSH_VERSION_ALREADY_EXIST>()) { |
1270 | 0 | LOG(WARNING) << "version already exist, version revert occurred. " |
1271 | 0 | << "tablet=" << _new_tablet->tablet_id() << ", version='" |
1272 | 0 | << rs_reader->version().first << "-" << rs_reader->version().second; |
1273 | 0 | _local_storage_engine.add_unused_rowset(new_rowset); |
1274 | 0 | have_failure_rowset = true; |
1275 | 0 | res = Status::OK(); |
1276 | 0 | } else if (!res) { |
1277 | 0 | LOG(WARNING) << "failed to register new version. " |
1278 | 0 | << " tablet=" << _new_tablet->tablet_id() |
1279 | 0 | << ", version=" << rs_reader->version().first << "-" |
1280 | 0 | << rs_reader->version().second; |
1281 | 0 | _local_storage_engine.add_unused_rowset(new_rowset); |
1282 | 0 | return process_alter_exit(); |
1283 | 0 | } else { |
1284 | 0 | VLOG_NOTICE << "register new version. tablet=" << _new_tablet->tablet_id() |
1285 | 0 | << ", version=" << rs_reader->version().first << "-" |
1286 | 0 | << rs_reader->version().second; |
1287 | 0 | } |
1288 | 0 | if (!have_failure_rowset) { |
1289 | 0 | *real_alter_version = rs_reader->version().second; |
1290 | 0 | } |
1291 | |
|
1292 | 0 | VLOG_TRACE << "succeed to convert a history version." |
1293 | 0 | << " version=" << rs_reader->version().first << "-" |
1294 | 0 | << rs_reader->version().second; |
1295 | 0 | } |
1296 | | |
1297 | | // XXX:The SchemaChange state should not be canceled at this time, because the new Delta has to be converted to the old and new Schema version |
1298 | 0 | return process_alter_exit(); |
1299 | 0 | } |
1300 | | |
1301 | | static const std::string WHERE_SIGN_LOWER = to_lower("__DORIS_WHERE_SIGN__"); |
1302 | | |
1303 | | // @static |
1304 | | // Analyze the mapping of the column and the mapping of the filter key |
1305 | | Status SchemaChangeJob::parse_request(const SchemaChangeParams& sc_params, |
1306 | | TabletSchema* base_tablet_schema, |
1307 | | TabletSchema* new_tablet_schema, BlockChanger* changer, |
1308 | 0 | bool* sc_sorting, bool* sc_directly) { |
1309 | 0 | changer->set_type(sc_params.alter_tablet_type); |
1310 | 0 | changer->set_compatible_version(sc_params.be_exec_version); |
1311 | |
|
1312 | 0 | const std::unordered_map<std::string, AlterMaterializedViewParam>& materialized_function_map = |
1313 | 0 | sc_params.materialized_params_map; |
1314 | 0 | DescriptorTbl desc_tbl = *sc_params.desc_tbl; |
1315 | | |
1316 | | // set column mapping |
1317 | 0 | for (int i = 0, new_schema_size = new_tablet_schema->num_columns(); i < new_schema_size; ++i) { |
1318 | 0 | const TabletColumn& new_column = new_tablet_schema->column(i); |
1319 | 0 | const std::string& column_name_lower = to_lower(new_column.name()); |
1320 | 0 | ColumnMapping* column_mapping = changer->get_mutable_column_mapping(i); |
1321 | 0 | column_mapping->new_column = &new_column; |
1322 | |
|
1323 | 0 | column_mapping->ref_column_idx = base_tablet_schema->field_index(new_column.name()); |
1324 | |
|
1325 | 0 | if (materialized_function_map.find(column_name_lower) != materialized_function_map.end()) { |
1326 | 0 | auto mv_param = materialized_function_map.find(column_name_lower)->second; |
1327 | 0 | column_mapping->expr = mv_param.expr; |
1328 | 0 | if (column_mapping->expr != nullptr) { |
1329 | 0 | continue; |
1330 | 0 | } |
1331 | 0 | } |
1332 | | |
1333 | 0 | if (column_mapping->ref_column_idx >= 0) { |
1334 | 0 | continue; |
1335 | 0 | } |
1336 | | |
1337 | 0 | if (sc_params.alter_tablet_type == ROLLUP) { |
1338 | 0 | std::string materialized_function_map_str; |
1339 | 0 | for (auto str : materialized_function_map) { |
1340 | 0 | if (!materialized_function_map_str.empty()) { |
1341 | 0 | materialized_function_map_str += ','; |
1342 | 0 | } |
1343 | 0 | materialized_function_map_str += str.first; |
1344 | 0 | } |
1345 | 0 | return Status::InternalError( |
1346 | 0 | "referenced column was missing. [column={},materialized_function_map={}]", |
1347 | 0 | new_column.name(), materialized_function_map_str); |
1348 | 0 | } |
1349 | | |
1350 | 0 | if (new_column.name().find("__doris_shadow_") == 0) { |
1351 | | // Should delete in the future, just a protection for bug. |
1352 | 0 | LOG(INFO) << "a shadow column is encountered " << new_column.name(); |
1353 | 0 | return Status::InternalError("failed due to operate on shadow column"); |
1354 | 0 | } |
1355 | | // Newly added column go here |
1356 | 0 | column_mapping->ref_column_idx = -1; |
1357 | |
|
1358 | 0 | if (i < base_tablet_schema->num_short_key_columns()) { |
1359 | 0 | *sc_directly = true; |
1360 | 0 | } |
1361 | 0 | RETURN_IF_ERROR( |
1362 | 0 | _init_column_mapping(column_mapping, new_column, new_column.default_value())); |
1363 | | |
1364 | 0 | LOG(INFO) << "A column with default value will be added after schema changing. " |
1365 | 0 | << "column=" << new_column.name() |
1366 | 0 | << ", default_value=" << new_column.default_value(); |
1367 | 0 | } |
1368 | | |
1369 | 0 | if (materialized_function_map.contains(WHERE_SIGN_LOWER)) { |
1370 | 0 | changer->set_where_expr(materialized_function_map.find(WHERE_SIGN_LOWER)->second.expr); |
1371 | 0 | } |
1372 | | |
1373 | | // If the reference sequence of the Key column is out of order, it needs to be reordered |
1374 | 0 | int num_default_value = 0; |
1375 | |
|
1376 | 0 | for (int i = 0, new_schema_size = new_tablet_schema->num_key_columns(); i < new_schema_size; |
1377 | 0 | ++i) { |
1378 | 0 | ColumnMapping* column_mapping = changer->get_mutable_column_mapping(i); |
1379 | |
|
1380 | 0 | if (!column_mapping->has_reference()) { |
1381 | 0 | num_default_value++; |
1382 | 0 | continue; |
1383 | 0 | } |
1384 | | |
1385 | 0 | if (column_mapping->ref_column_idx != i - num_default_value) { |
1386 | 0 | *sc_sorting = true; |
1387 | 0 | return Status::OK(); |
1388 | 0 | } |
1389 | 0 | } |
1390 | | |
1391 | 0 | if (base_tablet_schema->keys_type() != new_tablet_schema->keys_type()) { |
1392 | | // only when base table is dup and mv is agg |
1393 | | // the rollup job must be reagg. |
1394 | 0 | *sc_sorting = true; |
1395 | 0 | return Status::OK(); |
1396 | 0 | } |
1397 | | |
1398 | | // If the sort of key has not been changed but the new keys num is less then base's, |
1399 | | // the new table should be re agg. |
1400 | | // So we also need to set sc_sorting = true. |
1401 | | // A, B, C are keys(sort keys), D is value |
1402 | | // followings need resort: |
1403 | | // old keys: A B C D |
1404 | | // new keys: A B |
1405 | 0 | if (new_tablet_schema->keys_type() != KeysType::DUP_KEYS && |
1406 | 0 | new_tablet_schema->num_key_columns() < base_tablet_schema->num_key_columns()) { |
1407 | | // this is a table with aggregate key type, and num of key columns in new schema |
1408 | | // is less, which means the data in new tablet should be more aggregated. |
1409 | | // so we use sorting schema change to sort and merge the data. |
1410 | 0 | *sc_sorting = true; |
1411 | 0 | return Status::OK(); |
1412 | 0 | } |
1413 | | |
1414 | 0 | if (sc_params.alter_tablet_type == ROLLUP) { |
1415 | 0 | *sc_directly = true; |
1416 | 0 | return Status::OK(); |
1417 | 0 | } |
1418 | | |
1419 | 0 | if (sc_params.enable_unique_key_merge_on_write && |
1420 | 0 | new_tablet_schema->num_key_columns() > base_tablet_schema->num_key_columns()) { |
1421 | 0 | *sc_directly = true; |
1422 | 0 | return Status::OK(); |
1423 | 0 | } |
1424 | | |
1425 | 0 | if (base_tablet_schema->num_short_key_columns() != new_tablet_schema->num_short_key_columns()) { |
1426 | | // the number of short_keys changed, can't do linked schema change |
1427 | 0 | *sc_directly = true; |
1428 | 0 | return Status::OK(); |
1429 | 0 | } |
1430 | | |
1431 | 0 | if (!sc_params.delete_handler->empty()) { |
1432 | | // there exists delete condition in header, can't do linked schema change |
1433 | 0 | *sc_directly = true; |
1434 | 0 | return Status::OK(); |
1435 | 0 | } |
1436 | | |
1437 | | // if new tablet enable row store, or new tablet has different row store columns |
1438 | 0 | if ((!base_tablet_schema->exist_column(BeConsts::ROW_STORE_COL) && |
1439 | 0 | new_tablet_schema->exist_column(BeConsts::ROW_STORE_COL)) || |
1440 | 0 | !std::equal(new_tablet_schema->row_columns_uids().begin(), |
1441 | 0 | new_tablet_schema->row_columns_uids().end(), |
1442 | 0 | base_tablet_schema->row_columns_uids().begin(), |
1443 | 0 | base_tablet_schema->row_columns_uids().end())) { |
1444 | 0 | *sc_directly = true; |
1445 | 0 | } |
1446 | |
|
1447 | 0 | for (size_t i = 0; i < new_tablet_schema->num_columns(); ++i) { |
1448 | 0 | ColumnMapping* column_mapping = changer->get_mutable_column_mapping(i); |
1449 | 0 | if (column_mapping->expr != nullptr) { |
1450 | 0 | *sc_directly = true; |
1451 | 0 | return Status::OK(); |
1452 | 0 | } else if (column_mapping->ref_column_idx >= 0) { |
1453 | | // index changed |
1454 | 0 | if (vectorized::schema_util::has_schema_index_diff( |
1455 | 0 | new_tablet_schema, base_tablet_schema, i, column_mapping->ref_column_idx)) { |
1456 | 0 | *sc_directly = true; |
1457 | 0 | return Status::OK(); |
1458 | 0 | } |
1459 | 0 | } |
1460 | 0 | } |
1461 | | |
1462 | | // if rs_reader has remote files, link schema change is not supported, |
1463 | | // use directly schema change instead. |
1464 | 0 | if (!(*sc_directly) && !(*sc_sorting)) { |
1465 | | // check has remote rowset |
1466 | | // work for cloud and cold storage |
1467 | 0 | for (const auto& rs_reader : sc_params.ref_rowset_readers) { |
1468 | 0 | if (!rs_reader->rowset()->is_local()) { |
1469 | 0 | *sc_directly = true; |
1470 | 0 | break; |
1471 | 0 | } |
1472 | 0 | } |
1473 | 0 | } |
1474 | |
|
1475 | 0 | return Status::OK(); |
1476 | 0 | } |
1477 | | |
1478 | | Status SchemaChangeJob::_init_column_mapping(ColumnMapping* column_mapping, |
1479 | | const TabletColumn& column_schema, |
1480 | 0 | const std::string& value) { |
1481 | 0 | if (auto field = WrapperField::create(column_schema); field.has_value()) { |
1482 | 0 | column_mapping->default_value = field.value(); |
1483 | 0 | } else { |
1484 | 0 | return field.error(); |
1485 | 0 | } |
1486 | | |
1487 | 0 | if (column_schema.is_nullable() && value.length() == 0) { |
1488 | 0 | column_mapping->default_value->set_null(); |
1489 | 0 | } else { |
1490 | 0 | RETURN_IF_ERROR(column_mapping->default_value->from_string(value, column_schema.precision(), |
1491 | 0 | column_schema.frac())); |
1492 | 0 | } |
1493 | | |
1494 | 0 | return Status::OK(); |
1495 | 0 | } |
1496 | | |
1497 | 0 | Status SchemaChangeJob::_validate_alter_result(const TAlterTabletReqV2& request) { |
1498 | 0 | Version max_continuous_version = {-1, 0}; |
1499 | 0 | _new_tablet->max_continuous_version_from_beginning(&max_continuous_version); |
1500 | 0 | LOG(INFO) << "find max continuous version of tablet=" << _new_tablet->tablet_id() |
1501 | 0 | << ", start_version=" << max_continuous_version.first |
1502 | 0 | << ", end_version=" << max_continuous_version.second; |
1503 | 0 | if (max_continuous_version.second < request.alter_version) { |
1504 | 0 | return Status::InternalError("result version={} is less than request version={}", |
1505 | 0 | max_continuous_version.second, request.alter_version); |
1506 | 0 | } |
1507 | | |
1508 | 0 | std::vector<std::pair<Version, RowsetSharedPtr>> version_rowsets; |
1509 | 0 | { |
1510 | 0 | std::shared_lock rdlock(_new_tablet->get_header_lock()); |
1511 | 0 | _new_tablet->acquire_version_and_rowsets(&version_rowsets); |
1512 | 0 | } |
1513 | 0 | for (auto& pair : version_rowsets) { |
1514 | 0 | RowsetSharedPtr rowset = pair.second; |
1515 | 0 | if (!rowset->check_file_exist()) { |
1516 | 0 | return Status::Error<NOT_FOUND>( |
1517 | 0 | "SchemaChangeJob::_validate_alter_result meet invalid rowset"); |
1518 | 0 | } |
1519 | 0 | } |
1520 | 0 | return Status::OK(); |
1521 | 0 | } |
1522 | | |
1523 | | // For unique with merge-on-write table, should process delete bitmap here. |
1524 | | // 1. During double write, the newly imported rowsets does not calculate |
1525 | | // delete bitmap and publish successfully. |
1526 | | // 2. After conversion, calculate delete bitmap for the rowsets imported |
1527 | | // during double write. During this period, new data can still be imported |
1528 | | // witout calculating delete bitmap and publish successfully. |
1529 | | // 3. Block the new publish, calculate the delete bitmap of the |
1530 | | // incremental rowsets. |
1531 | | // 4. Switch the tablet status to TABLET_RUNNING. The newly imported |
1532 | | // data will calculate delete bitmap. |
1533 | 0 | Status SchemaChangeJob::_calc_delete_bitmap_for_mow_table(int64_t alter_version) { |
1534 | 0 | DBUG_EXECUTE_IF("SchemaChangeJob._calc_delete_bitmap_for_mow_table.random_failed", { |
1535 | 0 | if (rand() % 100 < (100 * dp->param("percent", 0.1))) { |
1536 | 0 | LOG_WARNING("SchemaChangeJob._calc_delete_bitmap_for_mow_table.random_failed"); |
1537 | 0 | return Status::InternalError("debug schema change calc delete bitmap random failed"); |
1538 | 0 | } |
1539 | 0 | }); |
1540 | | |
1541 | | // can't do compaction when calc delete bitmap, if the rowset being calculated does |
1542 | | // a compaction, it may cause the delete bitmap to be missed. |
1543 | 0 | std::lock_guard base_compaction_lock(_new_tablet->get_base_compaction_lock()); |
1544 | 0 | std::lock_guard cumu_compaction_lock(_new_tablet->get_cumulative_compaction_lock()); |
1545 | | |
1546 | | // step 2 |
1547 | 0 | int64_t max_version = _new_tablet->max_version().second; |
1548 | 0 | std::vector<RowsetSharedPtr> rowsets; |
1549 | 0 | if (alter_version < max_version) { |
1550 | 0 | LOG(INFO) << "alter table for unique with merge-on-write, calculate delete bitmap of " |
1551 | 0 | << "double write rowsets for version: " << alter_version + 1 << "-" << max_version |
1552 | 0 | << " new_tablet=" << _new_tablet->tablet_id(); |
1553 | 0 | std::shared_lock rlock(_new_tablet->get_header_lock()); |
1554 | 0 | RETURN_IF_ERROR(_new_tablet->capture_consistent_rowsets_unlocked( |
1555 | 0 | {alter_version + 1, max_version}, &rowsets)); |
1556 | 0 | } |
1557 | 0 | for (auto rowset_ptr : rowsets) { |
1558 | 0 | std::lock_guard rwlock(_new_tablet->get_rowset_update_lock()); |
1559 | 0 | std::shared_lock rlock(_new_tablet->get_header_lock()); |
1560 | 0 | RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(_new_tablet, rowset_ptr)); |
1561 | 0 | } |
1562 | | |
1563 | | // step 3 |
1564 | 0 | std::lock_guard rwlock(_new_tablet->get_rowset_update_lock()); |
1565 | 0 | std::lock_guard new_wlock(_new_tablet->get_header_lock()); |
1566 | 0 | SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
1567 | 0 | int64_t new_max_version = _new_tablet->max_version_unlocked(); |
1568 | 0 | rowsets.clear(); |
1569 | 0 | if (max_version < new_max_version) { |
1570 | 0 | LOG(INFO) << "alter table for unique with merge-on-write, calculate delete bitmap of " |
1571 | 0 | << "incremental rowsets for version: " << max_version + 1 << "-" |
1572 | 0 | << new_max_version << " new_tablet=" << _new_tablet->tablet_id(); |
1573 | 0 | RETURN_IF_ERROR(_new_tablet->capture_consistent_rowsets_unlocked( |
1574 | 0 | {max_version + 1, new_max_version}, &rowsets)); |
1575 | 0 | } |
1576 | 0 | for (auto&& rowset_ptr : rowsets) { |
1577 | 0 | RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(_new_tablet, rowset_ptr)); |
1578 | 0 | } |
1579 | | // step 4 |
1580 | 0 | RETURN_IF_ERROR(_new_tablet->set_tablet_state(TabletState::TABLET_RUNNING)); |
1581 | 0 | _new_tablet->save_meta(); |
1582 | 0 | return Status::OK(); |
1583 | 0 | } |
1584 | | |
1585 | | } // namespace doris |