diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp index a300cd39bcf5da..11f57919a31575 100644 --- a/be/src/cloud/cloud_meta_mgr.cpp +++ b/be/src/cloud/cloud_meta_mgr.cpp @@ -789,6 +789,12 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet, req.set_cumulative_point(tablet->cumulative_layer_point()); } req.set_end_version(-1); + if (tablet->tablet_state() == TABLET_NOTREADY) { + // Ask MS whether this NOT_READY tablet still has an active alter job, to + // tell an in-progress schema change new tablet from an abandoned shadow + // tablet, see CloudTablet::has_active_alter_job() + req.set_need_alter_job_info(true); + } VLOG_DEBUG << "send GetRowsetRequest: " << req.ShortDebugString(); // Host-level rate limiting for get_rowset @@ -863,6 +869,10 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet, int64_t now = duration_cast(system_clock::now().time_since_epoch()).count(); tablet->last_sync_time_s = now; + if (resp.has_has_alter_job()) { + tablet->set_has_active_alter_job(resp.has_alter_job()); + } + if (sync_stats) { sync_stats->get_remote_rowsets_rpc_ns += std::chrono::duration_cast(end - start).count(); diff --git a/be/src/cloud/cloud_storage_engine.h b/be/src/cloud/cloud_storage_engine.h index 330850b5ed8192..0df59452a8d366 100644 --- a/be/src/cloud/cloud_storage_engine.h +++ b/be/src/cloud/cloud_storage_engine.h @@ -156,6 +156,11 @@ class CloudStorageEngine final : public BaseStorageEngine { return _submitted_full_compactions.contains(tablet_id); } + bool is_preparing_cumu_compaction(int64_t tablet_id) const { + std::lock_guard lock(_compaction_mtx); + return _tablet_preparing_cumu_compaction.contains(tablet_id); + } + std::shared_ptr cumu_compaction_policy( std::string_view compaction_policy); diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp index d304358d950e41..c2b21d681e3058 100644 --- a/be/src/cloud/cloud_tablet.cpp +++ b/be/src/cloud/cloud_tablet.cpp @@ -39,6 +39,7 @@ #include #include "cloud/cloud_meta_mgr.h" +#include "cloud/cloud_cluster_info.h" #include "cloud/cloud_storage_engine.h" #include "cloud/cloud_tablet_mgr.h" #include "cloud/cloud_warm_up_manager.h" @@ -938,6 +939,62 @@ int64_t CloudTablet::get_cloud_cumu_compaction_score() const { return _approximate_cumu_num_deltas.load(std::memory_order_relaxed); } +// Keep the conditions consistent with the filter_out/disable/skip logic in +// CloudStorageEngine::_generate_cloud_compaction_tasks and +// CloudTabletMgr::get_topn_tablets_to_compact +std::string CloudTablet::compaction_not_scheduled_reason() { + if (is_zombie_tablet()) { + return "zombie"; + } + if (_tablet_meta->tablet_schema()->disable_auto_compaction()) { + return "disabled"; + } + if (_engine.has_base_compaction(tablet_id()) || _engine.has_cumu_compaction(tablet_id()) || + _engine.has_full_compaction(tablet_id()) || + _engine.is_preparing_cumu_compaction(tablet_id())) { + return "compaction_inflight"; + } + if (tablet_state() == TABLET_NOTREADY) { + if (!config::enable_new_tablet_do_compaction) { + return "alter_new_tablet_compaction_disabled"; + } + if (alter_version() == -1) { + return "waiting_alter_task"; + } + } + auto* cloud_cluster_info = + static_cast(ExecEnv::GetInstance()->cluster_info()); + if (config::enable_standby_passive_compaction && cloud_cluster_info->is_in_standby() && + fetch_add_approximate_num_rowsets(0) < + config::max_tablet_version_num * config::standby_compaction_version_ratio) { + return "standby_throttled"; + } + if (cloud_cluster_info->should_skip_compaction(this)) { + return "handled_by_other_cluster"; + } + int64_t now = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + auto cooldown_left = [now](int64_t last_ms) { + return last_ms + config::min_compaction_failure_interval_ms - now; + }; + if (int64_t ms = cooldown_left(last_cumu_compaction_failure_time()); ms > 0) { + return fmt::format("cumu_failure_cooldown({}ms)", ms); + } + if (int64_t ms = cooldown_left(last_cumu_no_suitable_version_ms); ms > 0) { + return fmt::format("cumu_no_suitable_version({}ms)", ms); + } + if (int64_t ms = cooldown_left(last_base_compaction_failure_time()); ms > 0) { + return fmt::format("base_failure_cooldown({}ms)", ms); + } + if (now - last_load_time_ms > config::compaction_load_max_freeze_interval_s * 1000 && + now - last_cumu_compaction_success_time_ms < config::cumu_compaction_interval_s * 1000 && + fetch_add_approximate_num_rowsets(0) < max_version_config() / 2) { + return "frozen"; + } + return ""; +} + // return a json string to show the compaction status of this tablet void CloudTablet::get_compaction_status(std::string* json_result) { rapidjson::Document root; @@ -970,6 +1027,21 @@ void CloudTablet::get_compaction_status(std::string* json_result) { cast_set(compaction_policy.length()), root.GetAllocator()); root.AddMember("compaction policy", compaction_policy_value, root.GetAllocator()); + rapidjson::Value tablet_state_value; + std::string tablet_state_str = tablet_state() == TABLET_RUNNING ? "RUNNING" : "NOTREADY"; + tablet_state_value.SetString(tablet_state_str.c_str(), + cast_set(tablet_state_str.length()), root.GetAllocator()); + root.AddMember("tablet state", tablet_state_value, root.GetAllocator()); + root.AddMember("is zombie tablet", is_zombie_tablet(), root.GetAllocator()); + root.AddMember("has active alter job", has_active_alter_job(), root.GetAllocator()); + root.AddMember("alter version", alter_version(), root.GetAllocator()); + root.AddMember("disable auto compaction", + _tablet_meta->tablet_schema()->disable_auto_compaction(), root.GetAllocator()); + rapidjson::Value reason_value; + std::string reason_str = compaction_not_scheduled_reason(); + reason_value.SetString(reason_str.c_str(), cast_set(reason_str.length()), + root.GetAllocator()); + root.AddMember("not scheduled reason", reason_value, root.GetAllocator()); root.AddMember("cumulative point", _cumulative_point.load(), root.GetAllocator()); rapidjson::Value cumu_value; std::string format_str = ToStringFromUnixMillis(_last_cumu_compaction_failure_millis.load()); diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h index 33a720f78fe0fc..454e792e86c1da 100644 --- a/be/src/cloud/cloud_tablet.h +++ b/be/src/cloud/cloud_tablet.h @@ -270,6 +270,32 @@ class CloudTablet final : public BaseTablet { int64_t alter_version() const { return _alter_version; } void set_alter_version(int64_t alter_version) { _alter_version = alter_version; } + // Whether this tablet still has an active alter (schema change) job registered + // in meta-service. Only meaningful for NOT_READY tablets; refreshed on each + // rowset sync (see GetRowsetResponse.has_alter_job). Defaults to true so that + // a tablet is treated as alter-in-progress until the first sync tells otherwise + // (also keeps behavior unchanged when MS has not been upgraded to fill it). + bool has_active_alter_job() const { + return _has_active_alter_job.load(std::memory_order_relaxed); + } + void set_has_active_alter_job(bool v) { + _has_active_alter_job.store(v, std::memory_order_relaxed); + } + + // A NOT_READY tablet whose schema change job has been cancelled or removed in + // meta-service. It will never be converted or compacted, does not serve reads + // or writes, and just stays in the local tablet cache until the recycler + // permanently deletes its meta. It should be invisible to compaction score + // stats, /api/compaction_score and compaction scheduling. + bool is_zombie_tablet() const { + return tablet_state() == TABLET_NOTREADY && !has_active_alter_job(); + } + + // Why this tablet is currently not scheduled for compaction, empty if it is + // schedulable. Display only (/api/compaction_score, /api/compaction/show), + // evaluated on demand, never in the scheduling loop. + std::string compaction_not_scheduled_reason(); + // Last active cluster info for compaction read-write separation std::string last_active_cluster_id() const { std::shared_lock lock(_cluster_info_mutex); @@ -497,6 +523,7 @@ class CloudTablet final : public BaseTablet { int64_t _max_version = -1; int64_t _base_size = 0; int64_t _alter_version = -1; + std::atomic _has_active_alter_job {true}; std::mutex _base_compaction_lock; std::mutex _cumulative_compaction_lock; diff --git a/be/src/cloud/cloud_tablet_mgr.cpp b/be/src/cloud/cloud_tablet_mgr.cpp index 82dd46a924dbdc..0ad1157510d9b3 100644 --- a/be/src/cloud/cloud_tablet_mgr.cpp +++ b/be/src/cloud/cloud_tablet_mgr.cpp @@ -499,6 +499,7 @@ Status CloudTabletMgr::get_topn_tablets_to_compact( auto disable = [](CloudTablet* t) { return t->tablet_meta()->tablet_schema()->disable_auto_compaction(); }; auto [num_filtered, num_disabled, num_skipped] = std::make_tuple(0, 0, 0); + int num_zombie = 0; auto weak_tablets = get_weak_tablets(); std::vector, int64_t>> buf; @@ -507,6 +508,11 @@ Status CloudTabletMgr::get_topn_tablets_to_compact( auto t = weak_tablet.lock(); if (t == nullptr) { continue; } + // Abandoned shadow tablets of cancelled schema change jobs will never be + // converted or compacted, exclude them from both the max score metrics and + // compaction candidates so that they don't masquerade as real backlog. + if (t->is_zombie_tablet()) { ++num_zombie; continue; } + int64_t s = score(t.get()); if (s <= 0) { continue; } if (s > score_stats->max_score) { @@ -535,6 +541,7 @@ Status CloudTabletMgr::get_topn_tablets_to_compact( LOG_EVERY_N(INFO, 1000) << "get_topn_compaction_score, n=" << n << " type=" << compaction_type << " num_tablets=" << weak_tablets.size() << " num_skipped=" << num_skipped << " num_disabled=" << num_disabled << " num_filtered=" << num_filtered + << " num_zombie=" << num_zombie << " max_score=" << score_stats->max_score << " max_score_tablet=" << max_score_tablet_id << " tablets=[" << [&buf] { std::stringstream ss; for (auto& i : buf) ss << i.first->tablet_id() << ":" << i.second << ","; return ss.str(); }() << "]" ; diff --git a/be/src/service/http/action/compaction_score_action.cpp b/be/src/service/http/action/compaction_score_action.cpp index 04febc7e1a470f..8e5fb05a90d3b4 100644 --- a/be/src/service/http/action/compaction_score_action.cpp +++ b/be/src/service/http/action/compaction_score_action.cpp @@ -57,6 +57,8 @@ namespace doris { const std::string TOP_N = "top_n"; const std::string SYNC_META = "sync_meta"; const std::string COMPACTION_SCORE = "compaction_score"; +const std::string TABLET_STATE = "tablet_state"; +const std::string NOT_SCHEDULED_REASON = "not_scheduled_reason"; constexpr size_t DEFAULT_TOP_N = std::numeric_limits::max(); constexpr bool DEFAULT_SYNC_META = false; @@ -73,7 +75,9 @@ std::vector calculate_compaction_scores( std::ranges::transform(tablets, std::back_inserter(result), [](const std::shared_ptr& tablet) -> CompactionScoreResult { return {.tablet_id = tablet->tablet_id(), - .compaction_score = tablet->get_real_compaction_score()}; + .compaction_score = tablet->get_real_compaction_score(), + .tablet_state = {}, + .not_scheduled_reason = {}}; }); return result; } @@ -99,6 +103,29 @@ struct CloudCompactionScoresAccessor final : CompactionScoresAccessor { return calculate_compaction_scores(s); } + // Only evaluated for the top-n entries about to be returned, so the cost is + // bounded by `top_n` regardless of how many tablets this BE caches. + void fill_details(std::span entries) override { + std::unordered_map by_id; + by_id.reserve(entries.size()); + for (auto& entry : entries) { + by_id.emplace(entry.tablet_id, &entry); + } + for (auto& weak_tablet : tablet_mgr.get_weak_tablets()) { + auto tablet = weak_tablet.lock(); + if (tablet == nullptr) { + continue; + } + auto it = by_id.find(tablet->tablet_id()); + if (it == by_id.end()) { + continue; + } + it->second->tablet_state = + tablet->tablet_state() == TABLET_RUNNING ? "RUNNING" : "NOTREADY"; + it->second->not_scheduled_reason = tablet->compaction_not_scheduled_reason(); + } + } + Status sync_meta() { auto tablets = get_all_tablets(); LOG(INFO) << "start to sync meta from ms"; @@ -122,8 +149,16 @@ struct CloudCompactionScoresAccessor final : CompactionScoresAccessor { std::vector tablets; tablets.reserve(weak_tablets.size()); for (auto& weak_tablet : weak_tablets) { + // Include RUNNING tablets and in-progress schema change new tablets, so + // that the result is consistent with what the compaction scheduler (and + // the max compaction score metric) sees. Abandoned shadow tablets of + // cancelled schema change jobs are excluded, see + // CloudTablet::is_zombie_tablet() if (auto tablet = weak_tablet.lock(); - tablet != nullptr and tablet->tablet_state() == TABLET_RUNNING) { + tablet != nullptr and + (tablet->tablet_state() == TABLET_RUNNING || + (tablet->tablet_state() == TABLET_NOTREADY && + tablet->has_active_alter_job()))) { tablets.push_back(std::move(tablet)); } } @@ -153,6 +188,24 @@ static rapidjson::Value jsonfy_tablet_compaction_score( node.AddMember(score_key, score_val, allocator); node.AddMember(tablet_id_key, tablet_id_val, allocator); + + if (!result.tablet_state.empty()) { + rapidjson::Value state_key; + state_key.SetString(TABLET_STATE.data(), cast_set(TABLET_STATE.size()), allocator); + rapidjson::Value state_val; + state_val.SetString(result.tablet_state.c_str(), + cast_set(result.tablet_state.length()), allocator); + node.AddMember(state_key, state_val, allocator); + } + if (!result.not_scheduled_reason.empty()) { + rapidjson::Value reason_key; + reason_key.SetString(NOT_SCHEDULED_REASON.data(), + cast_set(NOT_SCHEDULED_REASON.size()), allocator); + rapidjson::Value reason_val; + reason_val.SetString(result.not_scheduled_reason.c_str(), + cast_set(result.not_scheduled_reason.length()), allocator); + node.AddMember(reason_key, reason_val, allocator); + } return node; } @@ -220,6 +273,7 @@ Status CompactionScoreAction::_handle(size_t top_n, bool sync_meta, std::string* auto scores = _accessor->get_all_tablet_compaction_scores(); top_n = std::min(top_n, scores.size()); std::partial_sort(scores.begin(), scores.begin() + top_n, scores.end(), std::greater<>()); + _accessor->fill_details({scores.begin(), scores.begin() + top_n}); rapidjson::Document root; root.SetArray(); diff --git a/be/src/service/http/action/compaction_score_action.h b/be/src/service/http/action/compaction_score_action.h index 0fc88ae7627560..d0fc1f6c63ac3c 100644 --- a/be/src/service/http/action/compaction_score_action.h +++ b/be/src/service/http/action/compaction_score_action.h @@ -21,6 +21,7 @@ #include #include +#include #include #include "cloud/cloud_tablet_mgr.h" @@ -34,6 +35,13 @@ namespace doris { struct CompactionScoreResult { int64_t tablet_id; size_t compaction_score; + // Only filled in cloud mode: "RUNNING" or "NOTREADY" (an in-progress schema + // change new tablet; abandoned shadow tablets are excluded from the result) + std::string tablet_state; + // Only filled in cloud mode: why this tablet is currently not scheduled for + // compaction (empty if it is schedulable), e.g. "disabled", + // "compaction_inflight", "cumu_failure_cooldown(12000ms)" + std::string not_scheduled_reason; }; inline bool operator>(const CompactionScoreResult& lhs, const CompactionScoreResult& rhs) { @@ -44,6 +52,10 @@ struct CompactionScoresAccessor { virtual ~CompactionScoresAccessor() = default; virtual std::vector get_all_tablet_compaction_scores() = 0; + + // Fill display-only details (tablet state, not-scheduled reason) for the top-n + // entries about to be returned. Default is a no-op (local mode). + virtual void fill_details(std::span entries) {} }; // topn, sync diff --git a/be/test/cloud/cloud_compaction_test.cpp b/be/test/cloud/cloud_compaction_test.cpp index f435997e5de89a..0e3cbd8e3a2307 100644 --- a/be/test/cloud/cloud_compaction_test.cpp +++ b/be/test/cloud/cloud_compaction_test.cpp @@ -1423,4 +1423,59 @@ TEST_F(CloudCompactionTest, test_apply_txn_size_truncation_and_log_single_large_ ASSERT_EQ(truncated, 0); ASSERT_EQ(compaction.get_input_rowsets().size(), 1); } + +TEST_F(CloudCompactionTest, zombie_tablet_excluded_from_score_and_candidates) { + auto filter_out = [](CloudTablet* t) { return false; }; + CloudTabletMgr mgr(_engine); + + auto make_tablet = [&](int64_t tablet_id, TabletState state, bool has_alter_job, + int64_t cumu_deltas) { + TabletMetaSharedPtr meta(new TabletMeta(1, 2, 15673, 15674, 4, 5, TTabletSchema(), 6, + {{7, 8}}, UniqueId(9, 10), + TTabletType::TABLET_TYPE_DISK, + TCompressionType::LZ4F)); + meta->_tablet_id = tablet_id; + meta->set_tablet_state(state); + auto tablet = std::make_shared(_engine, meta); + tablet->set_has_active_alter_job(has_alter_job); + tablet->_approximate_cumu_num_deltas = cumu_deltas; + tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false); + mgr.put_tablet_for_UT(tablet); + return tablet; + }; + + // A normal running tablet with a low score + auto running = make_tablet(20001, TABLET_RUNNING, true, 3); + // An in-progress schema change new tablet: NOT_READY with an active alter job + auto live_shadow = make_tablet(20002, TABLET_NOTREADY, true, 50); + // An abandoned shadow tablet: NOT_READY whose alter job has been cancelled + auto zombie = make_tablet(20003, TABLET_NOTREADY, false, 551); + + ASSERT_FALSE(live_shadow->is_zombie_tablet()); + ASSERT_TRUE(zombie->is_zombie_tablet()); + + CompactionScoreStats score_stats; + std::vector> tablets; + Status st = mgr.get_topn_tablets_to_compact(10, CompactionType::CUMULATIVE_COMPACTION, + filter_out, &tablets, &score_stats); + ASSERT_EQ(st, Status::OK()); + // The zombie contributes to neither the max score nor the candidates, while + // the in-progress schema change new tablet still counts + ASSERT_EQ(score_stats.max_score, 50); + ASSERT_EQ(tablets.size(), 2); + ASSERT_EQ(tablets[0]->tablet_id(), 20002); + for (auto& t : tablets) { + ASSERT_NE(t->tablet_id(), 20003); + } + + // Once the alter job of the live shadow is gone as well, it becomes a zombie + // too and the max score falls back to the running tablet + live_shadow->set_has_active_alter_job(false); + st = mgr.get_topn_tablets_to_compact(10, CompactionType::CUMULATIVE_COMPACTION, filter_out, + &tablets, &score_stats); + ASSERT_EQ(st, Status::OK()); + ASSERT_EQ(score_stats.max_score, 3); + ASSERT_EQ(tablets.size(), 1); + ASSERT_EQ(tablets[0]->tablet_id(), 20001); +} } // namespace doris diff --git a/cloud/src/meta-service/meta_service.cpp b/cloud/src/meta-service/meta_service.cpp index c6aa71e38c8540..a1025e3d9db4e6 100644 --- a/cloud/src/meta-service/meta_service.cpp +++ b/cloud/src/meta-service/meta_service.cpp @@ -3355,6 +3355,30 @@ void MetaServiceImpl::get_rowset(::google::protobuf::RpcController* controller, } VLOG_DEBUG << "tablet_id=" << tablet_id << " stats=" << proto_to_json(tablet_stat); + if (request->need_alter_job_info()) { + // The requester caches this tablet in NOT_READY state. Report whether the + // tablet still has an active schema change job, so that BE can tell an + // in-progress schema change new tablet from an abandoned shadow tablet + // whose job has been cancelled or removed. The schema change job is + // registered under the new tablet's own job key when the job starts and + // cleared from it when the job is committed or aborted. + auto job_key = job_tablet_key({instance_id, idx.table_id(), idx.index_id(), + idx.partition_id(), tablet_id}); + std::string job_val; + TxnErrorCode job_err = txn->get(job_key, &job_val); + if (job_err != TxnErrorCode::TXN_OK && job_err != TxnErrorCode::TXN_KEY_NOT_FOUND) { + code = cast_as(job_err); + msg = fmt::format("failed to get tablet job, tablet_id={}, err={}", tablet_id, + job_err); + LOG(WARNING) << msg; + return; + } + TabletJobInfoPB job_pb; + response->set_has_alter_job(job_err == TxnErrorCode::TXN_OK && + job_pb.ParseFromString(job_val) && + job_pb.has_schema_change()); + } + int64_t bc_cnt = tablet_stat.base_compaction_cnt(); int64_t cc_cnt = tablet_stat.cumulative_compaction_cnt(); int64_t fc_cnt = diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto index 4addf9385b6457..4958aec845cd52 100644 --- a/gensrc/proto/cloud.proto +++ b/gensrc/proto/cloud.proto @@ -1446,6 +1446,10 @@ message GetRowsetRequest { optional SchemaOp schema_op = 8 [default = FILL_WITH_DICT]; optional string request_ip = 9; optional int64 full_compaction_cnt = 10; + // Set by BE when the locally cached tablet is in NOT_READY state, asking MS + // to additionally report whether the tablet still has an active alter job, + // see GetRowsetResponse.has_alter_job + optional bool need_alter_job_info = 11; } message GetRowsetResponse { @@ -1458,6 +1462,12 @@ message GetRowsetResponse { optional int64 partition_max_version = 5; // The cluster_id of the requester optional string requester_cluster_id = 6; + // Only filled when GetRowsetRequest.need_alter_job_info is set: whether the + // tablet still has an active schema change job registered in MS. A NOT_READY + // tablet without an alter job is an abandoned shadow tablet (its schema + // change job has been cancelled or removed) waiting to be recycled; it will + // never be converted or compacted. + optional bool has_alter_job = 7; } message GetSchemaDictRequest {