github-actions[bot] commented on code in PR #31055: URL: https://github.com/apache/doris/pull/31055#discussion_r1500456777
########## be/src/cloud/cloud_schema_change_job.cpp: ########## @@ -0,0 +1,360 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "cloud/cloud_schema_change_job.h" + +#include <gen_cpp/cloud.pb.h> + +#include <memory> + +#include "cloud/cloud_meta_mgr.h" +#include "cloud/cloud_tablet_mgr.h" +#include "common/status.h" +#include "olap/delete_handler.h" +#include "olap/rowset/beta_rowset.h" +#include "olap/rowset/rowset_factory.h" +#include "olap/rowset/segment_v2/inverted_index_desc.h" +#include "olap/storage_engine.h" +#include "olap/tablet.h" +#include "olap/tablet_fwd.h" +#include "olap/tablet_meta.h" +#include "service/backend_options.h" + +namespace doris { +using namespace ErrorCode; + +static constexpr int ALTER_TABLE_BATCH_SIZE = 4096; + +static std::unique_ptr<SchemaChange> get_sc_procedure(const BlockChanger& changer, + bool sc_sorting) { + if (sc_sorting) { + return std::make_unique<VBaseSchemaChangeWithSorting>( + changer, config::memory_limitation_per_thread_for_schema_change_bytes); + } + // else sc_directly + return std::make_unique<VSchemaChangeDirectly>(changer); +} + +CloudSchemaChangeJob::CloudSchemaChangeJob(CloudStorageEngine& cloud_storage_engine, + std::string job_id, int64_t expiration) + : _cloud_storage_engine(cloud_storage_engine), + _job_id(std::move(job_id)), + _expiration(expiration) {} + +CloudSchemaChangeJob::~CloudSchemaChangeJob() = default; + +Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { Review Comment: warning: function 'process_alter_tablet' exceeds recommended size/complexity thresholds [readability-function-size] ```cpp Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { ^ ``` <details> <summary>Additional context</summary> **be/src/cloud/cloud_schema_change_job.cpp:59:** 117 lines including whitespace and comments (threshold 80) ```cpp Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { ^ ``` </details> ########## be/src/cloud/cloud_schema_change_job.cpp: ########## @@ -0,0 +1,360 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "cloud/cloud_schema_change_job.h" + +#include <gen_cpp/cloud.pb.h> + +#include <memory> + +#include "cloud/cloud_meta_mgr.h" +#include "cloud/cloud_tablet_mgr.h" +#include "common/status.h" +#include "olap/delete_handler.h" +#include "olap/rowset/beta_rowset.h" +#include "olap/rowset/rowset_factory.h" +#include "olap/rowset/segment_v2/inverted_index_desc.h" +#include "olap/storage_engine.h" +#include "olap/tablet.h" +#include "olap/tablet_fwd.h" +#include "olap/tablet_meta.h" +#include "service/backend_options.h" + +namespace doris { +using namespace ErrorCode; + +static constexpr int ALTER_TABLE_BATCH_SIZE = 4096; + +static std::unique_ptr<SchemaChange> get_sc_procedure(const BlockChanger& changer, + bool sc_sorting) { + if (sc_sorting) { + return std::make_unique<VBaseSchemaChangeWithSorting>( + changer, config::memory_limitation_per_thread_for_schema_change_bytes); + } + // else sc_directly + return std::make_unique<VSchemaChangeDirectly>(changer); +} + +CloudSchemaChangeJob::CloudSchemaChangeJob(CloudStorageEngine& cloud_storage_engine, + std::string job_id, int64_t expiration) + : _cloud_storage_engine(cloud_storage_engine), + _job_id(std::move(job_id)), + _expiration(expiration) {} + +CloudSchemaChangeJob::~CloudSchemaChangeJob() = default; + +Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { + LOG(INFO) << "Begin to alter tablet. base_tablet_id=" << request.base_tablet_id + << ", new_tablet_id=" << request.new_tablet_id + << ", alter_version=" << request.alter_version << ", job_id=" << _job_id; + + // new tablet has to exist + _new_tablet = DORIS_TRY(_cloud_storage_engine.tablet_mgr().get_tablet(request.new_tablet_id)); + if (_new_tablet->tablet_state() == TABLET_RUNNING) { + LOG(INFO) << "schema change job has already finished. base_tablet_id=" + << request.base_tablet_id << ", new_tablet_id=" << request.new_tablet_id + << ", alter_version=" << request.alter_version << ", job_id=" << _job_id; + return Status::OK(); + } + + _base_tablet = DORIS_TRY(_cloud_storage_engine.tablet_mgr().get_tablet(request.base_tablet_id)); + + std::unique_lock<std::mutex> schema_change_lock(_base_tablet->get_schema_change_lock(), + std::try_to_lock); + if (!schema_change_lock.owns_lock()) { + LOG(WARNING) << "Failed to obtain schema change lock. base_tablet=" + << request.base_tablet_id; + return Status::Error<TRY_LOCK_FAILED>("Failed to obtain schema change lock. base_tablet={}", + request.base_tablet_id); + } + // MUST sync rowsets before capturing rowset readers and building DeleteHandler + RETURN_IF_ERROR(_base_tablet->sync_rowsets(request.alter_version)); + // ATTN: Only convert rowsets of version larger than 1, MUST let the new tablet cache have rowset [0-1] + _output_cumulative_point = _base_tablet->cumulative_layer_point(); + + std::vector<RowSetSplits> rs_splits; + int64_t base_max_version = _base_tablet->max_version_unlocked(); + if (request.alter_version > 1) { + // [0-1] is a placeholder rowset, no need to convert + RETURN_IF_ERROR(_base_tablet->capture_rs_readers({2, base_max_version}, &rs_splits, false)); + } + // FIXME(cyx): Should trigger compaction on base_tablet if there are too many rowsets to convert. + + // Create a new tablet schema, should merge with dropped columns in light weight schema change + _base_tablet_schema = std::make_shared<TabletSchema>(); + _base_tablet_schema->update_tablet_columns(*_base_tablet->tablet_schema(), request.columns); + _new_tablet_schema = _new_tablet->tablet_schema(); + + // delete handlers to filter out deleted rows + DeleteHandler delete_handler; + std::vector<RowsetMetaSharedPtr> delete_predicates; + for (auto& split : rs_splits) { + auto& rs_meta = split.rs_reader->rowset()->rowset_meta(); + if (rs_meta->has_delete_predicate()) { + _base_tablet_schema->merge_dropped_columns(*rs_meta->tablet_schema()); + delete_predicates.push_back(rs_meta); + } + } + RETURN_IF_ERROR(delete_handler.init(_base_tablet_schema, delete_predicates, base_max_version)); + + std::vector<ColumnId> return_columns; + return_columns.resize(_base_tablet_schema->num_columns()); + std::iota(return_columns.begin(), return_columns.end(), 0); + + // reader_context is stack variables, it's lifetime MUST keep the same with rs_readers + RowsetReaderContext reader_context; + reader_context.reader_type = ReaderType::READER_ALTER_TABLE; + reader_context.tablet_schema = _base_tablet_schema; + reader_context.need_ordered_result = true; + reader_context.delete_handler = &delete_handler; + reader_context.return_columns = &return_columns; + reader_context.sequence_id_idx = reader_context.tablet_schema->sequence_col_idx(); + reader_context.is_unique = _base_tablet->keys_type() == UNIQUE_KEYS; + reader_context.batch_size = ALTER_TABLE_BATCH_SIZE; + reader_context.delete_bitmap = &_base_tablet->tablet_meta()->delete_bitmap(); + reader_context.version = Version(0, base_max_version); + + for (auto& split : rs_splits) { + RETURN_IF_ERROR(split.rs_reader->init(&reader_context)); + } + + SchemaChangeParams sc_params; + + RETURN_IF_ERROR(DescriptorTbl::create(&sc_params.pool, request.desc_tbl, &sc_params.desc_tbl)); + sc_params.ref_rowset_readers.reserve(rs_splits.size()); + for (RowSetSplits& split : rs_splits) { + sc_params.ref_rowset_readers.emplace_back(std::move(split.rs_reader)); + } + sc_params.delete_handler = &delete_handler; + sc_params.be_exec_version = request.be_exec_version; + DCHECK(request.__isset.alter_tablet_type); + switch (request.alter_tablet_type) { + case TAlterTabletType::SCHEMA_CHANGE: + sc_params.alter_tablet_type = AlterTabletType::SCHEMA_CHANGE; + break; + case TAlterTabletType::ROLLUP: + sc_params.alter_tablet_type = AlterTabletType::ROLLUP; + break; + case TAlterTabletType::MIGRATION: + sc_params.alter_tablet_type = AlterTabletType::MIGRATION; + break; + } + if (!request.__isset.materialized_view_params) { + return _convert_historical_rowsets(sc_params); + } + for (auto item : request.materialized_view_params) { + AlterMaterializedViewParam mv_param; + mv_param.column_name = item.column_name; + /* + * origin_column_name is always be set now, + * but origin_column_name may be not set in some materialized view function. eg:count(1) + */ + if (item.__isset.origin_column_name) { + mv_param.origin_column_name = item.origin_column_name; + } + + if (item.__isset.mv_expr) { + mv_param.expr = std::make_shared<TExpr>(item.mv_expr); + } + sc_params.materialized_params_map.insert(std::make_pair(to_lower(item.column_name), mv_param)); + } + sc_params.enable_unique_key_merge_on_write = _new_tablet->enable_unique_key_merge_on_write(); + return _convert_historical_rowsets(sc_params); +} + +Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params) { Review Comment: warning: function '_convert_historical_rowsets' exceeds recommended size/complexity thresholds [readability-function-size] ```cpp Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params) { ^ ``` <details> <summary>Additional context</summary> **be/src/cloud/cloud_schema_change_job.cpp:178:** 180 lines including whitespace and comments (threshold 80) ```cpp Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params) { ^ ``` </details> ########## be/src/olap/schema_change.cpp: ########## @@ -934,80 +938,81 @@ } { std::lock_guard<std::shared_mutex> wrlock(_mutex); - _tablet_ids_in_converting.insert(new_tablet->tablet_id()); + _tablet_ids_in_converting.insert(_new_tablet->tablet_id()); } int64_t real_alter_version = 0; + sc_params.enable_unique_key_merge_on_write = + _new_tablet->enable_unique_key_merge_on_write(); res = _convert_historical_rowsets(sc_params, &real_alter_version); { std::lock_guard<std::shared_mutex> wrlock(_mutex); - _tablet_ids_in_converting.erase(new_tablet->tablet_id()); + _tablet_ids_in_converting.erase(_new_tablet->tablet_id()); } if (!res) { break; } - if (new_tablet->keys_type() == UNIQUE_KEYS && - new_tablet->enable_unique_key_merge_on_write()) { - res = _calc_delete_bitmap_for_mow_table(new_tablet, real_alter_version); + if (_new_tablet->keys_type() == UNIQUE_KEYS && + _new_tablet->enable_unique_key_merge_on_write()) { + res = _calc_delete_bitmap_for_mow_table(real_alter_version); if (!res) { break; } } else { // set state to ready - std::lock_guard<std::shared_mutex> new_wlock(new_tablet->get_header_lock()); + std::lock_guard<std::shared_mutex> new_wlock(_new_tablet->get_header_lock()); SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); - res = new_tablet->set_tablet_state(TabletState::TABLET_RUNNING); + res = _new_tablet->set_tablet_state(TabletState::TABLET_RUNNING); if (!res) { break; } - new_tablet->save_meta(); + _new_tablet->save_meta(); } } while (false); if (res) { // _validate_alter_result should be outside the above while loop. // to avoid requiring the header lock twice. - res = _validate_alter_result(new_tablet, request); + res = _validate_alter_result(request); } // if failed convert history data, then just remove the new tablet if (!res) { - LOG(WARNING) << "failed to alter tablet. base_tablet=" << base_tablet->tablet_id() - << ", drop new_tablet=" << new_tablet->tablet_id(); + LOG(WARNING) << "failed to alter tablet. base_tablet=" << _base_tablet->tablet_id() + << ", drop new_tablet=" << _new_tablet->tablet_id(); // do not drop the new tablet and its data. GC thread will } return res; } -bool SchemaChangeHandler::tablet_in_converting(int64_t tablet_id) { +bool SchemaChangeJob::tablet_in_converting(int64_t tablet_id) { std::shared_lock rdlock(_mutex); return _tablet_ids_in_converting.find(tablet_id) != _tablet_ids_in_converting.end(); } -Status SchemaChangeHandler::_get_versions_to_be_changed( - TabletSharedPtr base_tablet, std::vector<Version>* versions_to_be_changed, - RowsetSharedPtr* max_rowset) { - RowsetSharedPtr rowset = base_tablet->get_rowset_with_max_version(); +Status SchemaChangeJob::_get_versions_to_be_changed(std::vector<Version>* versions_to_be_changed, + RowsetSharedPtr* max_rowset) { + RowsetSharedPtr rowset = _base_tablet->get_rowset_with_max_version(); if (rowset == nullptr) { return Status::Error<ALTER_DELTA_DOES_NOT_EXISTS>("Tablet has no version. base_tablet={}", - base_tablet->tablet_id()); + _base_tablet->tablet_id()); } *max_rowset = rowset; - RETURN_IF_ERROR(base_tablet->capture_consistent_versions_unlocked( + RETURN_IF_ERROR(_base_tablet->capture_consistent_versions_unlocked( Version(0, rowset->version().second), versions_to_be_changed, false, false)); return Status::OK(); } // The `real_alter_version` parameter indicates that the version of [0-real_alter_version] is // converted from a base tablet, only used for the mow table now. -Status SchemaChangeHandler::_convert_historical_rowsets(const SchemaChangeParams& sc_params, - int64_t* real_alter_version) { +Status SchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params, Review Comment: warning: function '_convert_historical_rowsets' exceeds recommended size/complexity thresholds [readability-function-size] ```cpp Status SchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params, ^ ``` <details> <summary>Additional context</summary> **be/src/olap/schema_change.cpp:1010:** 129 lines including whitespace and comments (threshold 80) ```cpp Status SchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params, ^ ``` </details> ########## be/src/olap/schema_change.cpp: ########## @@ -652,97 +658,111 @@ return Status::OK(); } -Status SchemaChangeHandler::process_alter_tablet_v2(const TAlterTabletReqV2& request) { +Status VLocalSchemaChangeWithSorting::_inner_process(RowsetReaderSharedPtr rowset_reader, + RowsetWriter* rowset_writer, + BaseTabletSPtr new_tablet, + TabletSchemaSPtr base_tablet_schema, + TabletSchemaSPtr new_tablet_schema) { + Defer defer {[&]() { + // remove the intermediate rowsets generated by internal sorting + for (auto& row_set : _src_rowsets) { + _local_storage_engine.add_unused_rowset(row_set); + } + }}; + _pending_rs_guards.clear(); + LOG_INFO("lightman VLocalSchemaChangeWithSorting::_inner_process"); + return VBaseSchemaChangeWithSorting::_inner_process(rowset_reader, rowset_writer, new_tablet, + base_tablet_schema, new_tablet_schema); +} + +Status SchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { if (!request.__isset.desc_tbl) { return Status::Error<INVALID_ARGUMENT>( "desc_tbl is not set. Maybe the FE version is not equal to the BE " "version."); } + if (_base_tablet == nullptr) { + return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", + request.base_tablet_id); + } + if (_new_tablet == nullptr) { + return Status::Error<TABLE_NOT_FOUND>("fail to find new tablet. new_tablet={}", + request.new_tablet_id); + } LOG(INFO) << "begin to do request alter tablet: base_tablet_id=" << request.base_tablet_id << ", new_tablet_id=" << request.new_tablet_id << ", alter_version=" << request.alter_version; - TabletSharedPtr base_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.base_tablet_id); - if (base_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", - request.base_tablet_id); - } // Lock schema_change_lock util schema change info is stored in tablet header - std::unique_lock<std::mutex> schema_change_lock(base_tablet->get_schema_change_lock(), + std::unique_lock<std::mutex> schema_change_lock(_base_tablet->get_schema_change_lock(), std::try_to_lock); if (!schema_change_lock.owns_lock()) { return Status::Error<TRY_LOCK_FAILED>("failed to obtain schema change lock. base_tablet={}", request.base_tablet_id); } - Status res = _do_process_alter_tablet_v2(request); + Status res = _do_process_alter_tablet(request); LOG(INFO) << "finished alter tablet process, res=" << res; return res; } -std::shared_mutex SchemaChangeHandler::_mutex; -std::unordered_set<int64_t> SchemaChangeHandler::_tablet_ids_in_converting; +SchemaChangeJob::SchemaChangeJob(StorageEngine& local_storage_engine, + const TAlterTabletReqV2& request) + : _local_storage_engine(local_storage_engine) { + _base_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.base_tablet_id); + _new_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.new_tablet_id); + if (_base_tablet && _new_tablet) { + _base_tablet_schema = std::make_shared<TabletSchema>(); + _base_tablet_schema->update_tablet_columns(*_base_tablet->tablet_schema(), request.columns); + // During a schema change, the extracted columns of a variant should not be included in the tablet schema. + // This is because the schema change for a variant needs to ignore the extracted columns. + // Otherwise, the schema types in different rowsets might be inconsistent. When performing a schema change, + // the complete variant is constructed by reading all the sub-columns of the variant. + _new_tablet_schema = _new_tablet->tablet_schema()->copy_without_extracted_columns(); + } +} // In the past schema change and rollup will create new tablet and will wait for txns starting before the task to finished // It will cost a lot of time to wait and the task is very difficult to understand. // In alter task v2, FE will call BE to create tablet and send an alter task to BE to convert historical data. // The admin should upgrade all BE and then upgrade FE. // Should delete the old code after upgrade finished. -Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2& request) { - Status res = Status::OK(); - TabletSharedPtr base_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.base_tablet_id); - if (base_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", - request.base_tablet_id); - } - - signal::tablet_id = base_tablet->get_table_id(); - - // new tablet has to exist - TabletSharedPtr new_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.new_tablet_id); - if (new_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find new tablet. new_tablet={}", - request.new_tablet_id); - } +Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { Review Comment: warning: function '_do_process_alter_tablet' exceeds recommended size/complexity thresholds [readability-function-size] ```cpp Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { ^ ``` <details> <summary>Additional context</summary> **be/src/olap/schema_change.cpp:730:** 256 lines including whitespace and comments (threshold 80) ```cpp Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { ^ ``` </details> ########## be/src/io/cache/block/block_file_cache_factory.h: ########## @@ -46,6 +46,11 @@ class FileCacheFactory { size_t try_release(const std::string& base_path); + const std::string& get_cache_path() { Review Comment: warning: method 'get_cache_path' can be made static [readability-convert-member-functions-to-static] ```suggestion static const std::string& get_cache_path() { ``` ########## be/src/olap/schema_change.cpp: ########## @@ -1144,19 +1144,16 @@ // @static // Analyze the mapping of the column and the mapping of the filter key -Status SchemaChangeHandler::_parse_request(const SchemaChangeParams& sc_params, - BlockChanger* changer, bool* sc_sorting, - bool* sc_directly) { +Status SchemaChangeJob::parse_request(const SchemaChangeParams& sc_params, Review Comment: warning: function 'parse_request' exceeds recommended size/complexity thresholds [readability-function-size] ```cpp Status SchemaChangeJob::parse_request(const SchemaChangeParams& sc_params, ^ ``` <details> <summary>Additional context</summary> **be/src/olap/schema_change.cpp:1146:** 152 lines including whitespace and comments (threshold 80) ```cpp Status SchemaChangeJob::parse_request(const SchemaChangeParams& sc_params, ^ ``` </details> ########## be/src/olap/schema_change.cpp: ########## @@ -934,80 +938,81 @@ } { std::lock_guard<std::shared_mutex> wrlock(_mutex); - _tablet_ids_in_converting.insert(new_tablet->tablet_id()); + _tablet_ids_in_converting.insert(_new_tablet->tablet_id()); } int64_t real_alter_version = 0; + sc_params.enable_unique_key_merge_on_write = + _new_tablet->enable_unique_key_merge_on_write(); res = _convert_historical_rowsets(sc_params, &real_alter_version); { std::lock_guard<std::shared_mutex> wrlock(_mutex); - _tablet_ids_in_converting.erase(new_tablet->tablet_id()); + _tablet_ids_in_converting.erase(_new_tablet->tablet_id()); } if (!res) { break; } - if (new_tablet->keys_type() == UNIQUE_KEYS && - new_tablet->enable_unique_key_merge_on_write()) { - res = _calc_delete_bitmap_for_mow_table(new_tablet, real_alter_version); + if (_new_tablet->keys_type() == UNIQUE_KEYS && + _new_tablet->enable_unique_key_merge_on_write()) { + res = _calc_delete_bitmap_for_mow_table(real_alter_version); if (!res) { break; } } else { // set state to ready - std::lock_guard<std::shared_mutex> new_wlock(new_tablet->get_header_lock()); + std::lock_guard<std::shared_mutex> new_wlock(_new_tablet->get_header_lock()); SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); - res = new_tablet->set_tablet_state(TabletState::TABLET_RUNNING); + res = _new_tablet->set_tablet_state(TabletState::TABLET_RUNNING); if (!res) { break; } - new_tablet->save_meta(); + _new_tablet->save_meta(); } } while (false); if (res) { // _validate_alter_result should be outside the above while loop. // to avoid requiring the header lock twice. - res = _validate_alter_result(new_tablet, request); + res = _validate_alter_result(request); } // if failed convert history data, then just remove the new tablet if (!res) { - LOG(WARNING) << "failed to alter tablet. base_tablet=" << base_tablet->tablet_id() - << ", drop new_tablet=" << new_tablet->tablet_id(); + LOG(WARNING) << "failed to alter tablet. base_tablet=" << _base_tablet->tablet_id() + << ", drop new_tablet=" << _new_tablet->tablet_id(); // do not drop the new tablet and its data. GC thread will } return res; } -bool SchemaChangeHandler::tablet_in_converting(int64_t tablet_id) { +bool SchemaChangeJob::tablet_in_converting(int64_t tablet_id) { Review Comment: warning: method 'tablet_in_converting' can be made static [readability-convert-member-functions-to-static] be/src/olap/schema_change.h:281: ```diff - bool tablet_in_converting(int64_t tablet_id); + static bool tablet_in_converting(int64_t tablet_id); ``` ########## be/src/olap/schema_change.cpp: ########## @@ -652,97 +658,111 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_ return Status::OK(); } -Status SchemaChangeHandler::process_alter_tablet_v2(const TAlterTabletReqV2& request) { +Status VLocalSchemaChangeWithSorting::_inner_process(RowsetReaderSharedPtr rowset_reader, + RowsetWriter* rowset_writer, + BaseTabletSPtr new_tablet, + TabletSchemaSPtr base_tablet_schema, + TabletSchemaSPtr new_tablet_schema) { + Defer defer {[&]() { + // remove the intermediate rowsets generated by internal sorting + for (auto& row_set : _src_rowsets) { + _local_storage_engine.add_unused_rowset(row_set); + } + }}; + _pending_rs_guards.clear(); + LOG_INFO("lightman VLocalSchemaChangeWithSorting::_inner_process"); + return VBaseSchemaChangeWithSorting::_inner_process(rowset_reader, rowset_writer, new_tablet, + base_tablet_schema, new_tablet_schema); +} + +Status SchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) { if (!request.__isset.desc_tbl) { return Status::Error<INVALID_ARGUMENT>( "desc_tbl is not set. Maybe the FE version is not equal to the BE " "version."); } + if (_base_tablet == nullptr) { + return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", + request.base_tablet_id); + } + if (_new_tablet == nullptr) { + return Status::Error<TABLE_NOT_FOUND>("fail to find new tablet. new_tablet={}", + request.new_tablet_id); + } LOG(INFO) << "begin to do request alter tablet: base_tablet_id=" << request.base_tablet_id << ", new_tablet_id=" << request.new_tablet_id << ", alter_version=" << request.alter_version; - TabletSharedPtr base_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.base_tablet_id); - if (base_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", - request.base_tablet_id); - } // Lock schema_change_lock util schema change info is stored in tablet header - std::unique_lock<std::mutex> schema_change_lock(base_tablet->get_schema_change_lock(), + std::unique_lock<std::mutex> schema_change_lock(_base_tablet->get_schema_change_lock(), std::try_to_lock); if (!schema_change_lock.owns_lock()) { return Status::Error<TRY_LOCK_FAILED>("failed to obtain schema change lock. base_tablet={}", request.base_tablet_id); } - Status res = _do_process_alter_tablet_v2(request); + Status res = _do_process_alter_tablet(request); LOG(INFO) << "finished alter tablet process, res=" << res; return res; } -std::shared_mutex SchemaChangeHandler::_mutex; -std::unordered_set<int64_t> SchemaChangeHandler::_tablet_ids_in_converting; +SchemaChangeJob::SchemaChangeJob(StorageEngine& local_storage_engine, + const TAlterTabletReqV2& request) + : _local_storage_engine(local_storage_engine) { + _base_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.base_tablet_id); + _new_tablet = _local_storage_engine.tablet_manager()->get_tablet(request.new_tablet_id); + if (_base_tablet && _new_tablet) { + _base_tablet_schema = std::make_shared<TabletSchema>(); + _base_tablet_schema->update_tablet_columns(*_base_tablet->tablet_schema(), request.columns); + // During a schema change, the extracted columns of a variant should not be included in the tablet schema. + // This is because the schema change for a variant needs to ignore the extracted columns. + // Otherwise, the schema types in different rowsets might be inconsistent. When performing a schema change, + // the complete variant is constructed by reading all the sub-columns of the variant. + _new_tablet_schema = _new_tablet->tablet_schema()->copy_without_extracted_columns(); + } +} // In the past schema change and rollup will create new tablet and will wait for txns starting before the task to finished // It will cost a lot of time to wait and the task is very difficult to understand. // In alter task v2, FE will call BE to create tablet and send an alter task to BE to convert historical data. // The admin should upgrade all BE and then upgrade FE. // Should delete the old code after upgrade finished. -Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2& request) { - Status res = Status::OK(); - TabletSharedPtr base_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.base_tablet_id); - if (base_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find base tablet. base_tablet={}", - request.base_tablet_id); - } - - signal::tablet_id = base_tablet->get_table_id(); - - // new tablet has to exist - TabletSharedPtr new_tablet = - ExecEnv::GetInstance()->storage_engine().to_local().tablet_manager()->get_tablet( - request.new_tablet_id); - if (new_tablet == nullptr) { - return Status::Error<TABLE_NOT_FOUND>("fail to find new tablet. new_tablet={}", - request.new_tablet_id); - } +Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { Review Comment: warning: function '_do_process_alter_tablet' has cognitive complexity of 58 (threshold 50) [readability-function-cognitive-complexity] ```cpp Status SchemaChangeJob::_do_process_alter_tablet(const TAlterTabletReqV2& request) { ^ ``` <details> <summary>Additional context</summary> **be/src/olap/schema_change.cpp:736:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp if (_new_tablet->tablet_state() != TABLET_NOTREADY) { ^ ``` **be/src/olap/schema_change.cpp:757:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp if (!base_migration_rlock.owns_lock()) { ^ ``` **be/src/olap/schema_change.cpp:762:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp if (!new_migration_rlock.owns_lock()) { ^ ``` **be/src/olap/schema_change.cpp:784:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp for (int i = 0; i < num_cols; ++i) { ^ ``` **be/src/olap/schema_change.cpp:794:** nesting level increased to 1 ```cpp SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); ^ ``` **be/src/util/trace.h:32:** expanded from macro 'SCOPED_SIMPLE_TRACE_IF_TIMEOUT' ```cpp SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT(timeout, LOG(WARNING)) ^ ``` **be/src/util/trace.h:44:** expanded from macro 'SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT' ```cpp SCOPED_CLEANUP({ \ ^ ``` **be/src/util/scoped_cleanup.h:33:** expanded from macro 'SCOPED_CLEANUP' ```cpp auto VARNAME_LINENUM(scoped_cleanup) = MakeScopedCleanup([&] { func_body }); ^ ``` **be/src/olap/schema_change.cpp:794:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); ^ ``` **be/src/util/trace.h:32:** expanded from macro 'SCOPED_SIMPLE_TRACE_IF_TIMEOUT' ```cpp SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT(timeout, LOG(WARNING)) ^ ``` **be/src/util/trace.h:49:** expanded from macro 'SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT' ```cpp if (VARNAME_LINENUM(cost_us) >= VARNAME_LINENUM(timeout_us)) { \ ^ ``` **be/src/olap/schema_change.cpp:797:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp do { ^ ``` **be/src/olap/schema_change.cpp:800:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (!_get_versions_to_be_changed(&versions_to_be_changed, &max_rowset)) { ^ ``` **be/src/olap/schema_change.cpp:806:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) { ^ ``` **be/src/olap/schema_change.cpp:806:** +1 ```cpp if (max_rowset == nullptr || max_rowset->end_version() < request.alter_version) { ^ ``` **be/src/olap/schema_change.cpp:809:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp (max_rowset == nullptr ? 0 : max_rowset->end_version()), ^ ``` **be/src/olap/schema_change.cpp:839:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp RETURN_IF_ERROR(_new_tablet->modify_rowsets(empty_vec, rowsets_to_delete)); ^ ``` **be/src/common/status.h:541:** expanded from macro 'RETURN_IF_ERROR' ```cpp do { \ ^ ``` **be/src/olap/schema_change.cpp:839:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp RETURN_IF_ERROR(_new_tablet->modify_rowsets(empty_vec, rowsets_to_delete)); ^ ``` **be/src/common/status.h:543:** expanded from macro 'RETURN_IF_ERROR' ```cpp if (UNLIKELY(!_status_.ok())) { \ ^ ``` **be/src/olap/schema_change.cpp:856:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp RETURN_IF_ERROR( ^ ``` **be/src/common/status.h:541:** expanded from macro 'RETURN_IF_ERROR' ```cpp do { \ ^ ``` **be/src/olap/schema_change.cpp:856:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp RETURN_IF_ERROR( ^ ``` **be/src/common/status.h:543:** expanded from macro 'RETURN_IF_ERROR' ```cpp if (UNLIKELY(!_status_.ok())) { \ ^ ``` **be/src/olap/schema_change.cpp:858:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (rs_splits.empty()) { ^ ``` **be/src/olap/schema_change.cpp:874:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (!res) { ^ ``` **be/src/olap/schema_change.cpp:900:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp do { ^ ``` **be/src/olap/schema_change.cpp:901:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (!res) { ^ ``` **be/src/olap/schema_change.cpp:906:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp RETURN_IF_ERROR( ^ ``` **be/src/common/status.h:541:** expanded from macro 'RETURN_IF_ERROR' ```cpp do { \ ^ ``` **be/src/olap/schema_change.cpp:906:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp RETURN_IF_ERROR( ^ ``` **be/src/common/status.h:543:** expanded from macro 'RETURN_IF_ERROR' ```cpp if (UNLIKELY(!_status_.ok())) { \ ^ ``` **be/src/olap/schema_change.cpp:915:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp switch (request.alter_tablet_type) { ^ ``` **be/src/olap/schema_change.cpp:926:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (request.__isset.materialized_view_params) { ^ ``` **be/src/olap/schema_change.cpp:950:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (!res) { ^ ``` **be/src/olap/schema_change.cpp:954:** +2, including nesting penalty of 1, nesting level increased to 2 ```cpp if (_new_tablet->keys_type() == UNIQUE_KEYS && ^ ``` **be/src/olap/schema_change.cpp:957:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp if (!res) { ^ ``` **be/src/olap/schema_change.cpp:960:** +1, nesting level increased to 2 ```cpp } else { ^ ``` **be/src/olap/schema_change.cpp:963:** nesting level increased to 3 ```cpp SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); ^ ``` **be/src/util/trace.h:32:** expanded from macro 'SCOPED_SIMPLE_TRACE_IF_TIMEOUT' ```cpp SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT(timeout, LOG(WARNING)) ^ ``` **be/src/util/trace.h:44:** expanded from macro 'SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT' ```cpp SCOPED_CLEANUP({ \ ^ ``` **be/src/util/scoped_cleanup.h:33:** expanded from macro 'SCOPED_CLEANUP' ```cpp auto VARNAME_LINENUM(scoped_cleanup) = MakeScopedCleanup([&] { func_body }); ^ ``` **be/src/olap/schema_change.cpp:963:** +4, including nesting penalty of 3, nesting level increased to 4 ```cpp SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); ^ ``` **be/src/util/trace.h:32:** expanded from macro 'SCOPED_SIMPLE_TRACE_IF_TIMEOUT' ```cpp SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT(timeout, LOG(WARNING)) ^ ``` **be/src/util/trace.h:49:** expanded from macro 'SCOPED_SIMPLE_TRACE_TO_STREAM_IF_TIMEOUT' ```cpp if (VARNAME_LINENUM(cost_us) >= VARNAME_LINENUM(timeout_us)) { \ ^ ``` **be/src/olap/schema_change.cpp:965:** +3, including nesting penalty of 2, nesting level increased to 3 ```cpp if (!res) { ^ ``` **be/src/olap/schema_change.cpp:972:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp if (res) { ^ ``` **be/src/olap/schema_change.cpp:979:** +1, including nesting penalty of 0, nesting level increased to 1 ```cpp if (!res) { ^ ``` </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
