[fix](cloud) Exclude abandoned schema change shadow tablets from compaction score

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<seconds>(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<std::chrono::nanoseconds>(end - start).count();

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<CloudCumulativeCompactionPolicy> cumu_compaction_policy(
             std::string_view compaction_policy);
 

be/src/cloud/cloud_tablet.cpp

@@ -39,6 +39,7 @@
 #include <vector>
 
 #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<CloudClusterInfo*>(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::milliseconds>(
+                          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<uint32_t>(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<uint>(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<uint>(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());

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<bool> _has_active_alter_job {true};
 
     std::mutex _base_compaction_lock;
     std::mutex _cumulative_compaction_lock;

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<std::pair<std::shared_ptr<CloudTablet>, 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(); }() << "]"
                ;

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<size_t>::max();
 constexpr bool DEFAULT_SYNC_META = false;
 
@@ -73,7 +75,9 @@ std::vector<CompactionScoreResult> calculate_compaction_scores(
     std::ranges::transform(tablets, std::back_inserter(result),
                            [](const std::shared_ptr<T>& 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<CompactionScoreResult> entries) override {
+        std::unordered_map<int64_t, CompactionScoreResult*> 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<CloudTabletSPtr> 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<int32_t>(TABLET_STATE.size()), allocator);
+        rapidjson::Value state_val;
+        state_val.SetString(result.tablet_state.c_str(),
+                            cast_set<int32_t>(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<int32_t>(NOT_SCHEDULED_REASON.size()), allocator);
+        rapidjson::Value reason_val;
+        reason_val.SetString(result.not_scheduled_reason.c_str(),
+                             cast_set<int32_t>(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();

be/src/service/http/action/compaction_score_action.h

@@ -21,6 +21,7 @@
 
 #include <cstddef>
 #include <memory>
+#include <span>
 #include <string>
 
 #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<CompactionScoreResult> 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<CompactionScoreResult> entries) {}
 };
 
 // topn, sync

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<CloudTablet>(_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<std::shared_ptr<CloudTablet>> 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