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

Author: hust-hhb

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -371,40 +371,17 @@ Status CloudCumulativeCompaction::modify_rowsets() {
 Status 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
             DBUG_EXECUTE_IF("CloudCumulativeCompaction.modify_rowsets.update_delete_bitmap_failed",

be/src/cloud/cloud_delete_bitmap_action.cpp 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;

be/src/cloud/cloud_delete_bitmap_action.h 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

be/src/olap/compaction.cpp

@@ -854,6 +854,34 @@ void Compaction::construct_index_compaction_columns(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 &&

be/src/olap/compaction.h

@@ -84,6 +84,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;
 

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_total_size);
@@ -220,4 +224,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

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};
 };
 

be/src/olap/tablet.cpp

@@ -817,10 +817,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()) {
@@ -841,10 +844,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();
 

be/src/olap/tablet_meta.cpp

@@ -1216,7 +1216,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();

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"
@@ -377,6 +377,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",
@@ -425,9 +432,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