github-actions[bot] commented on code in PR #31055: URL: https://github.com/apache/doris/pull/31055#discussion_r1501981363
########## be/src/cloud/cloud_schema_change_job.cpp: ########## @@ -0,0 +1,361 @@ +// 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:** 118 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,361 @@ +// 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:179:** 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 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2& } { 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:** 128 lines including whitespace and comments (threshold 80) ```cpp Status SchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParams& sc_params, ^ ``` </details> ########## be/src/olap/schema_change.cpp: ########## @@ -1144,19 +1143,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:1145:** 152 lines including whitespace and comments (threshold 80) ```cpp Status SchemaChangeJob::parse_request(const SchemaChangeParams& sc_params, ^ ``` </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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
