platoneko commented on code in PR #31055:
URL: https://github.com/apache/doris/pull/31055#discussion_r1502482544
##########
be/src/olap/schema_change.cpp:
##########
@@ -1358,57 +1355,70 @@ Status
SchemaChangeHandler::_validate_alter_result(TabletSharedPtr new_tablet,
// incremental rowsets.
// 4. Switch the tablet status to TABLET_RUNNING. The newly imported
// data will calculate delete bitmap.
-Status SchemaChangeHandler::_calc_delete_bitmap_for_mow_table(TabletSharedPtr
new_tablet,
- int64_t
alter_version) {
-
DBUG_EXECUTE_IF("SchemaChangeHandler._calc_delete_bitmap_for_mow_table.random_failed",
{
+Status SchemaChangeJob::_calc_delete_bitmap_for_mow_table(int64_t
alter_version) {
+
DBUG_EXECUTE_IF("SchemaChangeJob._calc_delete_bitmap_for_mow_table.random_failed",
{
if (rand() % 100 < (100 * dp->param("percent", 0.1))) {
-
LOG_WARNING("SchemaChangeHandler._calc_delete_bitmap_for_mow_table.random_failed");
+
LOG_WARNING("SchemaChangeJob._calc_delete_bitmap_for_mow_table.random_failed");
return Status::InternalError("debug schema change calc delete
bitmap random failed");
}
});
// can't do compaction when calc delete bitmap, if the rowset being
calculated does
// a compaction, it may cause the delete bitmap to be missed.
- std::lock_guard
base_compaction_lock(new_tablet->get_base_compaction_lock());
- std::lock_guard
cumu_compaction_lock(new_tablet->get_cumulative_compaction_lock());
+ std::lock_guard
base_compaction_lock(_new_tablet->get_base_compaction_lock());
+ std::lock_guard
cumu_compaction_lock(_new_tablet->get_cumulative_compaction_lock());
// step 2
- int64_t max_version = new_tablet->max_version().second;
+ int64_t max_version = _new_tablet->max_version().second;
std::vector<RowsetSharedPtr> rowsets;
if (alter_version < max_version) {
LOG(INFO) << "alter table for unique with merge-on-write, calculate
delete bitmap of "
<< "double write rowsets for version: " << alter_version + 1
<< "-" << max_version
- << " new_tablet=" << new_tablet->tablet_id();
- std::shared_lock<std::shared_mutex>
rlock(new_tablet->get_header_lock());
- RETURN_IF_ERROR(new_tablet->capture_consistent_rowsets_unlocked(
+ << " new_tablet=" << _new_tablet->tablet_id();
+ std::shared_lock rlock(_new_tablet->get_header_lock());
+ RETURN_IF_ERROR(_new_tablet->capture_consistent_rowsets_unlocked(
{alter_version + 1, max_version}, &rowsets));
}
for (auto rowset_ptr : rowsets) {
- std::lock_guard<std::mutex>
rwlock(new_tablet->get_rowset_update_lock());
- std::shared_lock<std::shared_mutex>
rlock(new_tablet->get_header_lock());
- RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(new_tablet,
rowset_ptr));
+ std::lock_guard rwlock(_new_tablet->get_rowset_update_lock());
+ std::shared_lock rlock(_new_tablet->get_header_lock());
+ RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(_new_tablet,
rowset_ptr));
}
// step 3
- std::lock_guard<std::mutex> rwlock(new_tablet->get_rowset_update_lock());
- std::lock_guard<std::shared_mutex>
new_wlock(new_tablet->get_header_lock());
+ std::lock_guard rwlock(_new_tablet->get_rowset_update_lock());
+ std::lock_guard new_wlock(_new_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
- int64_t new_max_version = new_tablet->max_version_unlocked();
+ int64_t new_max_version = _new_tablet->max_version_unlocked();
rowsets.clear();
if (max_version < new_max_version) {
LOG(INFO) << "alter table for unique with merge-on-write, calculate
delete bitmap of "
<< "incremental rowsets for version: " << max_version + 1 <<
"-"
- << new_max_version << " new_tablet=" <<
new_tablet->tablet_id();
- RETURN_IF_ERROR(new_tablet->capture_consistent_rowsets_unlocked(
+ << new_max_version << " new_tablet=" <<
_new_tablet->tablet_id();
+ RETURN_IF_ERROR(_new_tablet->capture_consistent_rowsets_unlocked(
{max_version + 1, new_max_version}, &rowsets));
}
for (auto&& rowset_ptr : rowsets) {
- RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(new_tablet,
rowset_ptr));
+ RETURN_IF_ERROR(Tablet::update_delete_bitmap_without_lock(_new_tablet,
rowset_ptr));
}
// step 4
- RETURN_IF_ERROR(new_tablet->set_tablet_state(TabletState::TABLET_RUNNING));
- new_tablet->save_meta();
+
RETURN_IF_ERROR(_new_tablet->set_tablet_state(TabletState::TABLET_RUNNING));
+ _new_tablet->save_meta();
return Status::OK();
}
+Status execute_schema_change_job(const TAlterTabletReqV2& request) {
+ Status st;
+ if (config::is_cloud_mode()) {
+ DCHECK(request.__isset.job_id);
+ CloudSchemaChangeJob
job(ExecEnv::GetInstance()->storage_engine().to_cloud(),
+ std::to_string(request.job_id),
request.expiration);
+ st = job.process_alter_tablet(request);
+ } else {
+ SchemaChangeJob
job(ExecEnv::GetInstance()->storage_engine().to_local(), request);
+ st = job.process_alter_tablet(request);
+ }
Review Comment:
`cloud_start_workers` 可以拿到 `CloudStorageEngine`，看看怎么穿进来
--
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
