[improve](mow) merge and remove old version of delete bitmap when cumulative compaction is done

hust-hhb · hust-hhb · commit 73d9c5a3fda0 · 2024-10-10T10:32:29.000+08:00
diff --git a/be/src/cloud/cloud_cumulative_compaction.cpp b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -359,42 +359,18 @@ Status CloudCumulativeCompaction::modify_rowsets() {
 }
 
 void CloudCumulativeCompaction::process_old_version_delete_bitmap() {
-    // agg previously rowset old version delete bitmap
     std::vector<RowsetSharedPtr> pre_rowsets {};
-    std::vector<std::string> pre_rowset_ids {};
     for (const auto& it : cloud_tablet()->rowset_map()) {
         if (it.first.second < _input_rowsets.front()->start_version()) {
             pre_rowsets.emplace_back(it.second);
-            pre_rowset_ids.emplace_back(it.second->rowset_id().to_string());
         }
     }
     std::sort(pre_rowsets.begin(), pre_rowsets.end(), Rowset::comparator);
     if (!pre_rowsets.empty()) {
-        auto pre_max_version = _output_rowset->version().second;
-        DeleteBitmapPtr new_delete_bitmap =
-                std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id());
         std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey, DeleteBitmap::BitmapKey>>
                 to_remove_vec;
-        for (auto& rowset : pre_rowsets) {
-            if (rowset->rowset_meta()->total_disk_size() == 0) {
-                continue;
-            }
-            for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) {
-                rowset->rowset_id().to_string();
-                DeleteBitmap::BitmapKey start {rowset->rowset_id(), seg_id, 0};
-                DeleteBitmap::BitmapKey end {rowset->rowset_id(), seg_id, pre_max_version};
-                DeleteBitmap::BitmapKey before_end {rowset->rowset_id(), seg_id,
-                                                    pre_max_version - 1};
-                auto d = _tablet->tablet_meta()->delete_bitmap().get_agg(
-                        {rowset->rowset_id(), seg_id, pre_max_version});
-                to_remove_vec.emplace_back(
-                        std::make_tuple(_tablet->tablet_id(), start, before_end));
-                if (d->isEmpty()) {
-                    continue;
-                }
-                new_delete_bitmap->set(end, *d);
-            }
-        }
+        DeleteBitmapPtr new_delete_bitmap = nullptr;
+        agg_and_remove_old_version_delete_bitmap(pre_rowsets, to_remove_vec, new_delete_bitmap);
         if (!new_delete_bitmap->empty()) {
             // store agg delete bitmap
             Status update_st;
@@ -423,11 +399,11 @@ void CloudCumulativeCompaction::process_old_version_delete_bitmap() {
                 }
                 _tablet->tablet_meta()->delete_bitmap().add_to_remove_queue(version.to_string(),
                                                                             to_remove_vec);
-                DBUG_EXECUTE_IF(
-                        "CloudCumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets", {
-                            static_cast<CloudTablet*>(_tablet.get())
-                                    ->delete_expired_stale_rowsets();
-                        });
+                DBUG_EXECUTE_IF("CumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets",
+                                {
+                                    static_cast<CloudTablet*>(_tablet.get())
+                                            ->delete_expired_stale_rowsets();
+                                });
             }
         }
     }
diff --git a/be/src/http/action/delete_bitmap_action.cpp b/be/src/http/action/delete_bitmap_action.cpp
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "cloud_delete_bitmap_action.h"
+#include "delete_bitmap_action.h"
 
 #include <rapidjson/document.h>
 #include <rapidjson/encodings.h>
@@ -33,8 +33,10 @@
 #include <thread>
 #include <utility>
 
+#include "cloud/cloud_storage_engine.h"
 #include "cloud/cloud_tablet.h"
 #include "cloud/cloud_tablet_mgr.h"
+#include "cloud/config.h"
 #include "common/logging.h"
 #include "common/status.h"
 #include "gutil/strings/substitute.h"
@@ -43,7 +45,6 @@
 #include "http/http_request.h"
 #include "http/http_status.h"
 #include "olap/olap_define.h"
-#include "olap/storage_engine.h"
 #include "olap/tablet_manager.h"
 #include "util/doris_metrics.h"
 #include "util/stopwatch.hpp"
@@ -57,10 +58,9 @@ constexpr std::string_view HEADER_JSON = "application/json";
 
 } // namespace
 
-CloudDeleteBitmapAction::CloudDeleteBitmapAction(DeleteBitmapActionType ctype, ExecEnv* exec_env,
-                                                 CloudStorageEngine& engine,
-                                                 TPrivilegeHier::type hier,
-                                                 TPrivilegeType::type ptype)
+DeleteBitmapAction::DeleteBitmapAction(DeleteBitmapActionType ctype, ExecEnv* exec_env,
+                                       BaseStorageEngine& engine, TPrivilegeHier::type hier,
+                                       TPrivilegeType::type ptype)
         : HttpHandlerWithAuth(exec_env, hier, ptype),
           _engine(engine),
           _delete_bitmap_action_type(ctype) {}
@@ -78,20 +78,24 @@ static Status _check_param(HttpRequest* req, uint64_t* tablet_id) {
     return Status::OK();
 }
 
-Status CloudDeleteBitmapAction::_handle_show_delete_bitmap_count(HttpRequest* req,
-                                                                 std::string* json_result) {
+Status DeleteBitmapAction::_handle_show_delete_bitmap_count(HttpRequest* req,
+                                                            std::string* json_result) {
     uint64_t tablet_id = 0;
     // check & retrieve tablet_id from req if it contains
     RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id), "check param failed");
     if (tablet_id == 0) {
         return Status::InternalError("check param failed: missing tablet_id");
     }
 
-    CloudTabletSPtr tablet = DORIS_TRY(_engine.tablet_mgr().get_tablet(tablet_id));
+    BaseTabletSPtr tablet = nullptr;
+    if (config::is_cloud_mode()) {
+        tablet = DORIS_TRY(_engine.to_cloud().tablet_mgr().get_tablet(tablet_id));
+    } else {
+        tablet = _engine.to_local().tablet_manager()->get_tablet(tablet_id);
+    }
     if (tablet == nullptr) {
         return Status::NotFound("Tablet not found. tablet_id={}", tablet_id);
     }
-
     auto count = tablet->tablet_meta()->delete_bitmap().get_delete_bitmap_count();
     auto cardinality = tablet->tablet_meta()->delete_bitmap().cardinality();
     auto size = tablet->tablet_meta()->delete_bitmap().get_size();
@@ -111,7 +115,7 @@ Status CloudDeleteBitmapAction::_handle_show_delete_bitmap_count(HttpRequest* re
     return Status::OK();
 }
 
-void CloudDeleteBitmapAction::handle(HttpRequest* req) {
+void DeleteBitmapAction::handle(HttpRequest* req) {
     req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.data());
     if (_delete_bitmap_action_type == DeleteBitmapActionType::COUNT_INFO) {
         std::string json_result;
diff --git a/be/src/http/action/delete_bitmap_action.h b/be/src/http/action/delete_bitmap_action.h
@@ -21,9 +21,9 @@
 
 #include <string>
 
-#include "cloud/cloud_storage_engine.h"
 #include "common/status.h"
 #include "http/http_handler_with_auth.h"
+#include "olap/storage_engine.h"
 #include "olap/tablet.h"
 
 namespace doris {
@@ -34,21 +34,20 @@ class ExecEnv;
 enum class DeleteBitmapActionType { COUNT_INFO = 1 };
 
 /// This action is used for viewing the delete bitmap status
-class CloudDeleteBitmapAction : public HttpHandlerWithAuth {
+class DeleteBitmapAction : public HttpHandlerWithAuth {
 public:
-    CloudDeleteBitmapAction(DeleteBitmapActionType ctype, ExecEnv* exec_env,
-                            CloudStorageEngine& engine, TPrivilegeHier::type hier,
-                            TPrivilegeType::type ptype);
+    DeleteBitmapAction(DeleteBitmapActionType ctype, ExecEnv* exec_env, BaseStorageEngine& engine,
+                       TPrivilegeHier::type hier, TPrivilegeType::type ptype);
 
-    ~CloudDeleteBitmapAction() override = default;
+    ~DeleteBitmapAction() override = default;
 
     void handle(HttpRequest* req) override;
 
 private:
     Status _handle_show_delete_bitmap_count(HttpRequest* req, std::string* json_result);
 
 private:
-    CloudStorageEngine& _engine;
+    BaseStorageEngine& _engine;
     DeleteBitmapActionType _delete_bitmap_action_type;
 };
 } // namespace doris
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
@@ -900,6 +900,34 @@ void Compaction::construct_skip_inverted_index(RowsetWriterContext& ctx) {
     }
 }
 
+void Compaction::agg_and_remove_old_version_delete_bitmap(
+        std::vector<RowsetSharedPtr>& pre_rowsets,
+        std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey, DeleteBitmap::BitmapKey>>&
+                to_remove_vec,
+        DeleteBitmapPtr& new_delete_bitmap) {
+    // agg previously rowset old version delete bitmap
+    auto pre_max_version = _output_rowset->version().second;
+    new_delete_bitmap = std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id());
+    for (auto& rowset : pre_rowsets) {
+        if (rowset->rowset_meta()->total_disk_size() == 0) {
+            continue;
+        }
+        for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) {
+            rowset->rowset_id().to_string();
+            DeleteBitmap::BitmapKey start {rowset->rowset_id(), seg_id, 0};
+            DeleteBitmap::BitmapKey end {rowset->rowset_id(), seg_id, pre_max_version};
+            DeleteBitmap::BitmapKey before_end {rowset->rowset_id(), seg_id, pre_max_version - 1};
+            auto d = _tablet->tablet_meta()->delete_bitmap().get_agg(
+                    {rowset->rowset_id(), seg_id, pre_max_version});
+            to_remove_vec.emplace_back(std::make_tuple(_tablet->tablet_id(), start, before_end));
+            if (d->isEmpty()) {
+                continue;
+            }
+            new_delete_bitmap->set(end, *d);
+        }
+    }
+}
+
 Status CompactionMixin::construct_output_rowset_writer(RowsetWriterContext& ctx) {
     // only do index compaction for dup_keys and unique_keys with mow enabled
     if (config::inverted_index_compaction_enable &&
diff --git a/be/src/olap/compaction.h b/be/src/olap/compaction.h
@@ -83,6 +83,12 @@ class Compaction {
 
     int64_t merge_way_num();
 
+    void agg_and_remove_old_version_delete_bitmap(
+            std::vector<RowsetSharedPtr>& pre_rowsets,
+            std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey, DeleteBitmap::BitmapKey>>&
+                    to_remove_vec,
+            DeleteBitmapPtr& new_delete_bitmap);
+
     // the root tracker for this compaction
     std::shared_ptr<MemTrackerLimiter> _mem_tracker;
 
diff --git a/be/src/olap/cumulative_compaction.cpp b/be/src/olap/cumulative_compaction.cpp
@@ -124,6 +124,10 @@ Status CumulativeCompaction::execute_compact() {
           _output_rowset->num_segments() == 0)) {
         tablet()->set_last_cumu_compaction_success_time(UnixMillis());
     }
+    if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+        _tablet->enable_unique_key_merge_on_write() && _input_rowsets.size() != 1) {
+        process_old_version_delete_bitmap();
+    }
     DorisMetrics::instance()->cumulative_compaction_deltas_total->increment(_input_rowsets.size());
     DorisMetrics::instance()->cumulative_compaction_bytes_total->increment(_input_rowsets_size);
 
@@ -219,4 +223,34 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
     return Status::OK();
 }
 
+void CumulativeCompaction::process_old_version_delete_bitmap() {
+    std::vector<RowsetSharedPtr> pre_rowsets {};
+    for (const auto& it : tablet()->rowset_map()) {
+        if (it.first.second < _input_rowsets.front()->start_version()) {
+            pre_rowsets.emplace_back(it.second);
+        }
+    }
+    std::sort(pre_rowsets.begin(), pre_rowsets.end(), Rowset::comparator);
+    if (!pre_rowsets.empty()) {
+        std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey, DeleteBitmap::BitmapKey>>
+                to_remove_vec;
+        DeleteBitmapPtr new_delete_bitmap = nullptr;
+        agg_and_remove_old_version_delete_bitmap(pre_rowsets, to_remove_vec, new_delete_bitmap);
+        if (!new_delete_bitmap->empty()) {
+            // store agg delete bitmap
+            Version version(_input_rowsets.front()->start_version(),
+                            _input_rowsets.back()->end_version());
+            for (auto it = new_delete_bitmap->delete_bitmap.begin();
+                 it != new_delete_bitmap->delete_bitmap.end(); it++) {
+                _tablet->tablet_meta()->delete_bitmap().set(it->first, it->second);
+            }
+            _tablet->tablet_meta()->delete_bitmap().add_to_remove_queue(version.to_string(),
+                                                                        to_remove_vec);
+            DBUG_EXECUTE_IF("CumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets", {
+                static_cast<Tablet*>(_tablet.get())->delete_expired_stale_rowset();
+            });
+        }
+    }
+}
+
 } // namespace doris
diff --git a/be/src/olap/cumulative_compaction.h b/be/src/olap/cumulative_compaction.h
@@ -47,6 +47,8 @@ class CumulativeCompaction final : public CompactionMixin {
     void find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,
                                           std::vector<Version>* missing_version);
 
+    void process_old_version_delete_bitmap();
+
     Version _last_delete_version {-1, -1};
 };
 
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
@@ -796,10 +796,13 @@ void Tablet::delete_expired_stale_rowset() {
         auto old_meta_size = _tablet_meta->all_stale_rs_metas().size();
 
         // do delete operation
+        std::vector<std::string> version_to_delete;
         auto to_delete_iter = stale_version_path_map.begin();
         while (to_delete_iter != stale_version_path_map.end()) {
             std::vector<TimestampedVersionSharedPtr>& to_delete_version =
                     to_delete_iter->second->timestamped_versions();
+            int start_version = -1;
+            int end_version = -1;
             for (auto& timestampedVersion : to_delete_version) {
                 auto it = _stale_rs_version_map.find(timestampedVersion->version());
                 if (it != _stale_rs_version_map.end()) {
@@ -820,10 +823,17 @@ void Tablet::delete_expired_stale_rowset() {
                                  << timestampedVersion->version().second
                                  << "] not find in stale rs version map";
                 }
+                if (start_version < 0) {
+                    start_version = timestampedVersion->version().first;
+                }
+                end_version = timestampedVersion->version().second;
                 _delete_stale_rowset_by_version(timestampedVersion->version());
             }
+            Version version(start_version, end_version);
+            version_to_delete.emplace_back(version.to_string());
             to_delete_iter++;
         }
+        _tablet_meta->delete_bitmap().remove_stale_delete_bitmap_from_queue(version_to_delete);
 
         bool reconstructed = _reconstruct_version_tracker_if_necessary();
 
diff --git a/be/src/olap/tablet_meta.cpp b/be/src/olap/tablet_meta.cpp
@@ -1208,7 +1208,7 @@ void DeleteBitmap::remove_stale_delete_bitmap_from_queue(const std::vector<std::
             _stale_delete_bitmap.erase(version_str);
         }
     }
-    if (tablet_id == -1 || to_delete.empty()) {
+    if (tablet_id == -1 || to_delete.empty() || !config::is_cloud_mode()) {
         return;
     }
     CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();
diff --git a/be/src/service/http_service.cpp b/be/src/service/http_service.cpp
@@ -25,7 +25,6 @@
 #include <vector>
 
 #include "cloud/cloud_compaction_action.h"
-#include "cloud/cloud_delete_bitmap_action.h"
 #include "cloud/config.h"
 #include "cloud/injection_point_action.h"
 #include "common/config.h"
@@ -42,6 +41,7 @@
 #include "http/action/compaction_score_action.h"
 #include "http/action/config_action.h"
 #include "http/action/debug_point_action.h"
+#include "http/action/delete_bitmap_action.h"
 #include "http/action/download_action.h"
 #include "http/action/download_binlog_action.h"
 #include "http/action/file_cache_action.h"
@@ -364,6 +364,13 @@ void HttpService::register_local_handler(StorageEngine& engine) {
 
     _ev_http_server->register_handler(HttpMethod::GET, "/api/compaction/run_status",
                                       run_status_compaction_action);
+
+    DeleteBitmapAction* count_delete_bitmap_action =
+            _pool.add(new DeleteBitmapAction(DeleteBitmapActionType::COUNT_INFO, _env, engine,
+                                             TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
+    _ev_http_server->register_handler(HttpMethod::GET, "/api/delete_bitmap/count",
+                                      count_delete_bitmap_action);
+
     CheckTabletSegmentAction* check_tablet_segment_action = _pool.add(new CheckTabletSegmentAction(
             _env, engine, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
     _ev_http_server->register_handler(HttpMethod::POST, "/api/check_tablet_segment_lost",
@@ -412,9 +419,9 @@ void HttpService::register_cloud_handler(CloudStorageEngine& engine) {
                                       TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
     _ev_http_server->register_handler(HttpMethod::GET, "/api/compaction/run_status",
                                       run_status_compaction_action);
-    CloudDeleteBitmapAction* count_delete_bitmap_action =
-            _pool.add(new CloudDeleteBitmapAction(DeleteBitmapActionType::COUNT_INFO, _env, engine,
-                                                  TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
+    DeleteBitmapAction* count_delete_bitmap_action =
+            _pool.add(new DeleteBitmapAction(DeleteBitmapActionType::COUNT_INFO, _env, engine,
+                                             TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
     _ev_http_server->register_handler(HttpMethod::GET, "/api/delete_bitmap/count",
                                       count_delete_bitmap_action);
 #ifdef ENABLE_INJECTION_POINT
diff --git a/regression-test/suites/compaction/test_cu_compaction_remove_old_version_delete_bitmap.groovy b/regression-test/suites/compaction/test_cu_compaction_remove_old_version_delete_bitmap.groovy

Original file line number	Diff line number	Diff line change
`@@ -1208,7 +1208,7 @@ void DeleteBitmap::remove_stale_delete_bitmap_from_queue(const std::vector<std::`
`1208`	`1208`	`_stale_delete_bitmap.erase(version_str);`
`1209`	`1209`	`}`
`1210`	`1210`	`}`
`1211`		`- if (tablet_id == -1 \|\| to_delete.empty()) {`
	`1211`	`+ if (tablet_id == -1 \|\| to_delete.empty() \|\| !config::is_cloud_mode()) {`
`1212`	`1212`	`return;`
`1213`	`1213`	`}`
`1214`	`1214`	`CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();`