Coverage Report

Created: 2025-03-13 18:46

/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