This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 76818506d92 [improve](cloud-mow) merge and remove old version of
delete bitmap when cumulative compaction is done (#40204)
76818506d92 is described below
commit 76818506d92b61e7e7e727a6f8eb993f643fe4f1
Author: huanghaibin <284824...@qq.com>
AuthorDate: Fri Sep 20 16:51:29 2024 +0800
[improve](cloud-mow) merge and remove old version of delete bitmap when
cumulative compaction is done (#40204)
When the count of delete bitmap is big, it may lead to high cpu use
rate, we can merge old version delete bitmap to reduce the count of
delete bitmap when cumulative compaction is done to reduce the use rate
of cpu.
Here is an example:
before cu compaciton ,rowset layout is
"cumulative point": 10,
"rowsets": [
"[0-1] 0 DATA NONOVERLAPPING
0200000000000000ffffffffffffffecffffffffffffffff 0",
"[2-9] 1 DATA NONOVERLAPPING
0200000000002cc1a8447a4e0ba5850f773803ae2d534284 1.21 KB",
"[10-10] 1 DATA NONOVERLAPPING
0200000000002d07a8447a4e0ba5850f773803ae2d534284 1.09 KB",
"[11-11] 1 DATA NONOVERLAPPING
0200000000002d14a8447a4e0ba5850f773803ae2d534284 1.10 KB",
"[12-12] 1 DATA NONOVERLAPPING
0200000000002d23a8447a4e0ba5850f773803ae2d534284 1.10 KB",
"[13-13] 1 DATA NONOVERLAPPING
0200000000002d33a8447a4e0ba5850f773803ae2d534284 1.10 KB",
"[14-14] 1 DATA NONOVERLAPPING
0200000000002d40a8447a4e0ba5850f773803ae2d534284 1.10 KB"
],
base rowset is [2-9], now cu compaciton input rowset range is [10-14],
after compaciton, rowset layout is
"cumulative point": 10,
"rowsets": [
"[0-1] 0 DATA NONOVERLAPPING
0200000000000000ffffffffffffffecffffffffffffffff 0",
"[2-9] 1 DATA NONOVERLAPPING
0200000000002cc1a8447a4e0ba5850f773803ae2d534284 1.21 KB",
"[10-14] 1 DATA NONOVERLAPPING
0200000000002d63a8447a4e0ba5850f773803ae2d534284 1.21 KB"
],
1. the delete bitmap of rowset [10-10] -> [11-11] -> [12-12] -> [13-13]
-> [14-14] will be deleted when delete_expired_stale_rowsets
2. the delete bitmap of version 10~14, will do agg first, then update
the agg result on verison 14 delete bitmap
3. if update sucess, add the delete bitmap of version 10~13 to the queue
which name is to_remove_vec
4. when deleting delete bitmap on rowset [10-10] -> [11-11] -> [12-12]
-> [13-13] -> [14-14], the delete bitmap which mention on step 3 will be
deleted too.
---
be/src/cloud/cloud_cumulative_compaction.cpp | 85 +++++-
be/src/cloud/cloud_cumulative_compaction.h | 2 +
.../cloud/cloud_cumulative_compaction_policy.cpp | 7 +-
be/src/cloud/cloud_cumulative_compaction_policy.h | 5 +-
be/src/cloud/cloud_delete_bitmap_action.cpp | 127 +++++++++
be/src/cloud/cloud_delete_bitmap_action.h | 54 ++++
.../cloud/cloud_engine_calc_delete_bitmap_task.cpp | 2 +
be/src/cloud/cloud_meta_mgr.cpp | 51 +++-
be/src/cloud/cloud_meta_mgr.h | 7 +
be/src/cloud/cloud_storage_engine.cpp | 2 +-
be/src/cloud/cloud_tablet.cpp | 11 +-
be/src/cloud/config.cpp | 2 +
be/src/cloud/config.h | 2 +
be/src/olap/rowset/beta_rowset_writer.cpp | 3 +-
be/src/olap/tablet_meta.cpp | 58 ++++
be/src/olap/tablet_meta.h | 18 ++
be/src/service/http_service.cpp | 6 +
cloud/src/common/bvars.cpp | 1 +
cloud/src/common/bvars.h | 1 +
cloud/src/meta-service/meta_service.cpp | 108 +++++--
cloud/src/meta-service/meta_service.h | 12 +
cloud/src/meta-service/meta_service_helper.h | 3 +
gensrc/proto/cloud.proto | 14 +
...compaction_remove_old_version_delete_bitmap.out | 29 ++
...paction_remove_old_version_delete_bitmap.groovy | 316 +++++++++++++++++++++
25 files changed, 891 insertions(+), 35 deletions(-)
diff --git a/be/src/cloud/cloud_cumulative_compaction.cpp
b/be/src/cloud/cloud_cumulative_compaction.cpp
index 29cfe412fa8..8cee815843c 100644
--- a/be/src/cloud/cloud_cumulative_compaction.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -18,6 +18,7 @@
#include "cloud/cloud_cumulative_compaction.h"
#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
#include "common/config.h"
#include "common/logging.h"
@@ -27,6 +28,7 @@
#include "olap/compaction.h"
#include "olap/cumulative_compaction_policy.h"
#include "service/backend_options.h"
+#include "util/debug_points.h"
#include "util/trace.h"
#include "util/uuid_generator.h"
@@ -254,10 +256,10 @@ Status CloudCumulativeCompaction::modify_rowsets() {
compaction_job->add_output_rowset_ids(_output_rowset->rowset_id().to_string());
DeleteBitmapPtr output_rowset_delete_bitmap = nullptr;
+ int64_t initiator =
+ HashUtil::hash64(_uuid.data(), _uuid.size(), 0) &
std::numeric_limits<int64_t>::max();
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write()) {
- int64_t initiator = HashUtil::hash64(_uuid.data(), _uuid.size(), 0) &
- std::numeric_limits<int64_t>::max();
RETURN_IF_ERROR(cloud_tablet()->calc_delete_bitmap_for_compaction(
_input_rowsets, _output_rowset, _rowid_conversion,
compaction_type(),
_stats.merged_rows, initiator, output_rowset_delete_bitmap,
@@ -340,9 +342,88 @@ Status CloudCumulativeCompaction::modify_rowsets() {
stats.num_rows(),
stats.data_size());
}
}
+ if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ _tablet->enable_unique_key_merge_on_write() && _input_rowsets.size()
!= 1) {
+ process_old_version_delete_bitmap();
+ }
return Status::OK();
}
+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);
+ }
+ }
+ if (!new_delete_bitmap->empty()) {
+ // store agg delete bitmap
+ Status update_st;
+
DBUG_EXECUTE_IF("CloudCumulativeCompaction.modify_rowsets.update_delete_bitmap_failed",
+ {
+ update_st = Status::InternalError(
+ "test fail to update delete bitmap for
tablet_id {}",
+ cloud_tablet()->tablet_id());
+ });
+ if (update_st.ok()) {
+ update_st =
_engine.meta_mgr().update_delete_bitmap_without_lock(
+ *cloud_tablet(), new_delete_bitmap.get());
+ }
+ if (!update_st.ok()) {
+ std::stringstream ss;
+ ss << "failed to update delete bitmap for tablet=" <<
cloud_tablet()->tablet_id()
+ << " st=" << update_st.to_string();
+ std::string msg = ss.str();
+ LOG(WARNING) << msg;
+ } else {
+ 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(
+
"CloudCumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets", {
+ static_cast<CloudTablet*>(_tablet.get())
+ ->delete_expired_stale_rowsets();
+ });
+ }
+ }
+ }
+}
+
void CloudCumulativeCompaction::garbage_collection() {
CloudCompactionMixin::garbage_collection();
cloud::TabletJobInfoPB job;
diff --git a/be/src/cloud/cloud_cumulative_compaction.h
b/be/src/cloud/cloud_cumulative_compaction.h
index f353d0f5189..62c7cb44ea5 100644
--- a/be/src/cloud/cloud_cumulative_compaction.h
+++ b/be/src/cloud/cloud_cumulative_compaction.h
@@ -47,6 +47,8 @@ private:
void update_cumulative_point();
+ void process_old_version_delete_bitmap();
+
ReaderType compaction_type() const override { return
ReaderType::READER_CUMULATIVE_COMPACTION; }
std::string _uuid;
diff --git a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
index b8c4ee20cb2..f9af469e56f 100644
--- a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
@@ -34,12 +34,11 @@ namespace doris {
CloudSizeBasedCumulativeCompactionPolicy::CloudSizeBasedCumulativeCompactionPolicy(
int64_t promotion_size, double promotion_ratio, int64_t
promotion_min_size,
- int64_t compaction_min_size, int64_t promotion_version_count)
+ int64_t compaction_min_size)
: _promotion_size(promotion_size),
_promotion_ratio(promotion_ratio),
_promotion_min_size(promotion_min_size),
- _compaction_min_size(compaction_min_size),
- _promotion_version_count(promotion_version_count) {}
+ _compaction_min_size(compaction_min_size) {}
int64_t CloudSizeBasedCumulativeCompactionPolicy::_level_size(const int64_t
size) {
if (size < 1024) return 0;
@@ -205,7 +204,7 @@ int64_t
CloudSizeBasedCumulativeCompactionPolicy::new_cumulative_point(
// consider it's version count here.
bool satisfy_promotion_version =
tablet->enable_unique_key_merge_on_write() &&
output_rowset->end_version() -
output_rowset->start_version() >
- _promotion_version_count;
+
config::compaction_promotion_version_count;
// if rowsets have delete version, move to the last directly.
// if rowsets have no delete version, check output_rowset total disk size
satisfies promotion size.
return output_rowset->start_version() == last_cumulative_point &&
diff --git a/be/src/cloud/cloud_cumulative_compaction_policy.h
b/be/src/cloud/cloud_cumulative_compaction_policy.h
index 66068e9f3be..c142a8a6d3d 100644
--- a/be/src/cloud/cloud_cumulative_compaction_policy.h
+++ b/be/src/cloud/cloud_cumulative_compaction_policy.h
@@ -59,8 +59,7 @@ public:
int64_t promotion_size = config::compaction_promotion_size_mbytes
* 1024 * 1024,
double promotion_ratio = config::compaction_promotion_ratio,
int64_t promotion_min_size =
config::compaction_promotion_min_size_mbytes * 1024 * 1024,
- int64_t compaction_min_size = config::compaction_min_size_mbytes *
1024 * 1024,
- int64_t promotion_version_count =
config::compaction_promotion_version_count);
+ int64_t compaction_min_size = config::compaction_min_size_mbytes *
1024 * 1024);
~CloudSizeBasedCumulativeCompactionPolicy() override = default;
@@ -94,8 +93,6 @@ private:
int64_t _promotion_min_size;
/// lower bound size to do compaction compaction.
int64_t _compaction_min_size;
- // cumulative compaction promotion version count, only works for unique
key MoW table
- int64_t _promotion_version_count;
};
class CloudTimeSeriesCumulativeCompactionPolicy : public
CloudCumulativeCompactionPolicy {
diff --git a/be/src/cloud/cloud_delete_bitmap_action.cpp
b/be/src/cloud/cloud_delete_bitmap_action.cpp
new file mode 100644
index 00000000000..60db5896dfa
--- /dev/null
+++ b/be/src/cloud/cloud_delete_bitmap_action.cpp
@@ -0,0 +1,127 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "cloud_delete_bitmap_action.h"
+
+#include <rapidjson/document.h>
+#include <rapidjson/encodings.h>
+#include <rapidjson/prettywriter.h>
+#include <rapidjson/rapidjson.h>
+#include <rapidjson/stringbuffer.h>
+
+#include <chrono> // IWYU pragma: keep
+#include <exception>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <sstream>
+#include <string>
+#include <thread>
+#include <utility>
+
+#include "cloud/cloud_tablet.h"
+#include "cloud/cloud_tablet_mgr.h"
+#include "common/logging.h"
+#include "common/status.h"
+#include "gutil/strings/substitute.h"
+#include "http/http_channel.h"
+#include "http/http_headers.h"
+#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"
+
+namespace doris {
+using namespace ErrorCode;
+
+namespace {
+
+constexpr std::string_view HEADER_JSON = "application/json";
+
+} // namespace
+
+CloudDeleteBitmapAction::CloudDeleteBitmapAction(DeleteBitmapActionType ctype,
ExecEnv* exec_env,
+ CloudStorageEngine& engine,
+ TPrivilegeHier::type hier,
+ TPrivilegeType::type ptype)
+ : HttpHandlerWithAuth(exec_env, hier, ptype),
+ _engine(engine),
+ _delete_bitmap_action_type(ctype) {}
+
+static Status _check_param(HttpRequest* req, uint64_t* tablet_id) {
+ const auto& req_tablet_id = req->param(TABLET_ID_KEY);
+ if (req_tablet_id.empty()) {
+ return Status::InternalError("tablet id is empty!");
+ }
+ try {
+ *tablet_id = std::stoull(req_tablet_id);
+ } catch (const std::exception& e) {
+ return Status::InternalError("convert tablet_id failed, {}", e.what());
+ }
+ return Status::OK();
+}
+
+Status CloudDeleteBitmapAction::_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));
+ 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();
+
+ rapidjson::Document root;
+ root.SetObject();
+ root.AddMember("delete_bitmap_count", count, root.GetAllocator());
+ root.AddMember("cardinality", cardinality, root.GetAllocator());
+ root.AddMember("size", size, root.GetAllocator());
+
+ // to json string
+ rapidjson::StringBuffer strbuf;
+ rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf);
+ root.Accept(writer);
+ *json_result = std::string(strbuf.GetString());
+
+ return Status::OK();
+}
+
+void CloudDeleteBitmapAction::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;
+ Status st = _handle_show_delete_bitmap_count(req, &json_result);
+ if (!st.ok()) {
+ HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());
+ } else {
+ HttpChannel::send_reply(req, HttpStatus::OK, json_result);
+ }
+ }
+}
+
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/cloud/cloud_delete_bitmap_action.h
b/be/src/cloud/cloud_delete_bitmap_action.h
new file mode 100644
index 00000000000..9321661374c
--- /dev/null
+++ b/be/src/cloud/cloud_delete_bitmap_action.h
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <stdint.h>
+
+#include <string>
+
+#include "cloud/cloud_storage_engine.h"
+#include "common/status.h"
+#include "http/http_handler_with_auth.h"
+#include "olap/tablet.h"
+
+namespace doris {
+class HttpRequest;
+
+class ExecEnv;
+
+enum class DeleteBitmapActionType { COUNT_INFO = 1 };
+
+/// This action is used for viewing the delete bitmap status
+class CloudDeleteBitmapAction : public HttpHandlerWithAuth {
+public:
+ CloudDeleteBitmapAction(DeleteBitmapActionType ctype, ExecEnv* exec_env,
+ CloudStorageEngine& engine, TPrivilegeHier::type
hier,
+ TPrivilegeType::type ptype);
+
+ ~CloudDeleteBitmapAction() override = default;
+
+ void handle(HttpRequest* req) override;
+
+private:
+ Status _handle_show_delete_bitmap_count(HttpRequest* req, std::string*
json_result);
+
+private:
+ CloudStorageEngine& _engine;
+ DeleteBitmapActionType _delete_bitmap_action_type;
+};
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
index 5c369af38a6..6abc3958650 100644
--- a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
+++ b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
@@ -87,6 +87,7 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
LOG(WARNING) << "handle calc delete bitmap fail, st=" <<
st.to_string();
}
});
+ VLOG_DEBUG << "submit TabletCalcDeleteBitmapTask for tablet=" <<
tablet_id;
if (!submit_st.ok()) {
_res = submit_st;
break;
@@ -126,6 +127,7 @@ void
CloudTabletCalcDeleteBitmapTask::set_compaction_stats(int64_t ms_base_compa
}
Status CloudTabletCalcDeleteBitmapTask::handle() const {
+ VLOG_DEBUG << "start calculate delete bitmap on tablet " << _tablet_id;
SCOPED_ATTACH_TASK(_mem_tracker);
int64_t t1 = MonotonicMicros();
auto base_tablet = DORIS_TRY(_engine.get_tablet(_tablet_id));
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 36217ddb61d..071c15d11e5 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -290,6 +290,8 @@ static std::string debug_info(const Request& req) {
return "";
} else if constexpr (is_any_v<Request, CreateRowsetRequest>) {
return fmt::format(" tablet_id={}", req.rowset_meta().tablet_id());
+ } else if constexpr (is_any_v<Request, RemoveDeleteBitmapRequest>) {
+ return fmt::format(" tablet_id={}", req.tablet_id());
} else {
static_assert(!sizeof(Request));
}
@@ -1046,16 +1048,43 @@ Status CloudMetaMgr::update_delete_bitmap(const
CloudTablet& tablet, int64_t loc
return st;
}
+Status CloudMetaMgr::update_delete_bitmap_without_lock(const CloudTablet&
tablet,
+ DeleteBitmap*
delete_bitmap) {
+ VLOG_DEBUG << "update_delete_bitmap_without_lock , tablet_id: " <<
tablet.tablet_id();
+ UpdateDeleteBitmapRequest req;
+ UpdateDeleteBitmapResponse res;
+ req.set_cloud_unique_id(config::cloud_unique_id);
+ req.set_table_id(tablet.table_id());
+ req.set_partition_id(tablet.partition_id());
+ req.set_tablet_id(tablet.tablet_id());
+ // use a fake lock id to resolve compatibility issues
+ req.set_lock_id(-3);
+ req.set_unlock(true);
+ for (auto& [key, bitmap] : delete_bitmap->delete_bitmap) {
+ req.add_rowset_ids(std::get<0>(key).to_string());
+ req.add_segment_ids(std::get<1>(key));
+ req.add_versions(std::get<2>(key));
+ // To save space, convert array and bitmap containers to run containers
+ bitmap.runOptimize();
+ std::string bitmap_data(bitmap.getSizeInBytes(), '\0');
+ bitmap.write(bitmap_data.data());
+ *(req.add_segment_delete_bitmaps()) = std::move(bitmap_data);
+ }
+ return retry_rpc("update delete bitmap", req, &res,
&MetaService_Stub::update_delete_bitmap);
+}
+
Status CloudMetaMgr::get_delete_bitmap_update_lock(const CloudTablet& tablet,
int64_t lock_id,
int64_t initiator) {
- VLOG_DEBUG << "get_delete_bitmap_update_lock , tablet_id: " <<
tablet.tablet_id();
+ VLOG_DEBUG << "get_delete_bitmap_update_lock , tablet_id: " <<
tablet.tablet_id()
+ << ",lock_id:" << lock_id;
GetDeleteBitmapUpdateLockRequest req;
GetDeleteBitmapUpdateLockResponse res;
req.set_cloud_unique_id(config::cloud_unique_id);
req.set_table_id(tablet.table_id());
req.set_lock_id(lock_id);
req.set_initiator(initiator);
- req.set_expiration(10); // 10s expiration time for compaction and
schema_change
+ // set expiration time for compaction and schema_change
+ req.set_expiration(config::delete_bitmap_lock_expiration_seconds);
int retry_times = 0;
Status st;
std::default_random_engine rng = make_random_engine();
@@ -1076,4 +1105,22 @@ Status CloudMetaMgr::get_delete_bitmap_update_lock(const
CloudTablet& tablet, in
return st;
}
+Status CloudMetaMgr::remove_old_version_delete_bitmap(
+ int64_t tablet_id,
+ const std::vector<std::tuple<std::string, uint64_t, uint64_t>>&
to_delete) {
+ LOG(INFO) << "remove_old_version_delete_bitmap , tablet_id: " << tablet_id;
+ RemoveDeleteBitmapRequest req;
+ RemoveDeleteBitmapResponse res;
+ req.set_cloud_unique_id(config::cloud_unique_id);
+ req.set_tablet_id(tablet_id);
+ for (auto& value : to_delete) {
+ req.add_rowset_ids(std::get<0>(value));
+ req.add_begin_versions(std::get<1>(value));
+ req.add_end_versions(std::get<2>(value));
+ }
+ auto st = retry_rpc("remove old delete bitmap", req, &res,
+ &MetaService_Stub::remove_delete_bitmap);
+ return st;
+}
+
} // namespace doris::cloud
diff --git a/be/src/cloud/cloud_meta_mgr.h b/be/src/cloud/cloud_meta_mgr.h
index 2f776b05686..79cdb3fd3d1 100644
--- a/be/src/cloud/cloud_meta_mgr.h
+++ b/be/src/cloud/cloud_meta_mgr.h
@@ -95,9 +95,16 @@ public:
Status update_delete_bitmap(const CloudTablet& tablet, int64_t lock_id,
int64_t initiator,
DeleteBitmap* delete_bitmap);
+ Status update_delete_bitmap_without_lock(const CloudTablet& tablet,
+ DeleteBitmap* delete_bitmap);
+
Status get_delete_bitmap_update_lock(const CloudTablet& tablet, int64_t
lock_id,
int64_t initiator);
+ Status remove_old_version_delete_bitmap(
+ int64_t tablet_id,
+ const std::vector<std::tuple<std::string, uint64_t, uint64_t>>&
to_delete);
+
private:
bool sync_tablet_delete_bitmap_by_cache(CloudTablet* tablet, int64_t
old_max_version,
std::ranges::range auto&& rs_metas,
diff --git a/be/src/cloud/cloud_storage_engine.cpp
b/be/src/cloud/cloud_storage_engine.cpp
index 1cf6e2ac039..4f452656a62 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -257,7 +257,7 @@ Status CloudStorageEngine::start_bg_threads() {
// add calculate tablet delete bitmap task thread pool
RETURN_IF_ERROR(ThreadPoolBuilder("TabletCalDeleteBitmapThreadPool")
- .set_min_threads(1)
+
.set_min_threads(config::calc_tablet_delete_bitmap_task_max_thread)
.set_max_threads(config::calc_tablet_delete_bitmap_task_max_thread)
.build(&_calc_tablet_delete_bitmap_task_thread_pool));
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index a7e482a7eb7..576f1da7262 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -364,6 +364,7 @@ int CloudTablet::delete_expired_stale_rowsets() {
std::vector<RowsetSharedPtr> expired_rowsets;
int64_t expired_stale_sweep_endtime =
::time(nullptr) - config::tablet_rowset_stale_sweep_time_sec;
+ std::vector<std::string> version_to_delete;
{
std::unique_lock wlock(_meta_lock);
@@ -376,6 +377,8 @@ int CloudTablet::delete_expired_stale_rowsets() {
}
for (int64_t path_id : path_ids) {
+ int start_version = -1;
+ int end_version = -1;
// delete stale versions in version graph
auto version_path =
_timestamped_version_tracker.fetch_and_delete_path_by_id(path_id);
for (auto& v_ts : version_path->timestamped_versions()) {
@@ -390,12 +393,18 @@ int CloudTablet::delete_expired_stale_rowsets() {
DCHECK(false) << [this, &wlock]() { wlock.unlock();
std::string json; get_compaction_status(&json); return json; }();
// clang-format on
}
+ if (start_version < 0) {
+ start_version = v_ts->version().first;
+ }
+ end_version = v_ts->version().second;
_tablet_meta->delete_stale_rs_meta_by_version(v_ts->version());
- VLOG_DEBUG << "delete stale rowset " << v_ts->version();
}
+ Version version(start_version, end_version);
+ version_to_delete.emplace_back(version.to_string());
}
_reconstruct_version_tracker_if_necessary();
}
+
_tablet_meta->delete_bitmap().remove_stale_delete_bitmap_from_queue(version_to_delete);
recycle_cached_data(expired_rowsets);
return expired_rowsets.size();
}
diff --git a/be/src/cloud/config.cpp b/be/src/cloud/config.cpp
index 3a93928fd0b..e724dbea84e 100644
--- a/be/src/cloud/config.cpp
+++ b/be/src/cloud/config.cpp
@@ -65,6 +65,8 @@ DEFINE_mInt32(sync_load_for_tablets_thread, "32");
DEFINE_mBool(enable_new_tablet_do_compaction, "false");
+DEFINE_Int32(delete_bitmap_lock_expiration_seconds, "10");
+
DEFINE_Bool(enable_cloud_txn_lazy_commit, "false");
DEFINE_mInt32(remove_expired_tablet_txn_info_interval_seconds, "300");
diff --git a/be/src/cloud/config.h b/be/src/cloud/config.h
index eb40cd09e31..86197f924d0 100644
--- a/be/src/cloud/config.h
+++ b/be/src/cloud/config.h
@@ -97,6 +97,8 @@ DECLARE_mBool(save_load_error_log_to_s3);
// the theads which sync the datas which loaded in other clusters
DECLARE_mInt32(sync_load_for_tablets_thread);
+DECLARE_Int32(delete_bitmap_lock_expiration_seconds);
+
// enable large txn lazy commit in meta-service `commit_txn`
DECLARE_mBool(enable_cloud_txn_lazy_commit);
diff --git a/be/src/olap/rowset/beta_rowset_writer.cpp
b/be/src/olap/rowset/beta_rowset_writer.cpp
index 45f260bdfa1..e3f657e4d07 100644
--- a/be/src/olap/rowset/beta_rowset_writer.cpp
+++ b/be/src/olap/rowset/beta_rowset_writer.cpp
@@ -282,7 +282,8 @@ Status
BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) {
LOG(INFO) << "[Memtable Flush] construct delete bitmap tablet: " <<
_context.tablet->tablet_id()
<< ", rowset_ids: " << _context.mow_context->rowset_ids.size()
<< ", cur max_version: " << _context.mow_context->max_version
- << ", transaction_id: " << _context.mow_context->txn_id
+ << ", transaction_id: " << _context.mow_context->txn_id << ",
delete_bitmap_count: "
+ <<
_context.tablet->tablet_meta()->delete_bitmap().get_delete_bitmap_count()
<< ", cost: " << watch.get_elapse_time_us() << "(us), total
rows: " << total_rows;
return Status::OK();
}
diff --git a/be/src/olap/tablet_meta.cpp b/be/src/olap/tablet_meta.cpp
index fbb924089b1..a0d6213860c 100644
--- a/be/src/olap/tablet_meta.cpp
+++ b/be/src/olap/tablet_meta.cpp
@@ -30,6 +30,8 @@
#include <set>
#include <utility>
+#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_storage_engine.h"
#include "cloud/config.h"
#include "common/config.h"
#include "gutil/integral_types.h"
@@ -1082,6 +1084,7 @@ bool DeleteBitmap::empty() const {
}
uint64_t DeleteBitmap::cardinality() const {
+ std::shared_lock l(lock);
uint64_t res = 0;
for (auto entry : delete_bitmap) {
res += entry.second.cardinality();
@@ -1089,6 +1092,15 @@ uint64_t DeleteBitmap::cardinality() const {
return res;
}
+size_t DeleteBitmap::get_size() const {
+ std::shared_lock l(lock);
+ size_t charge = 0;
+ for (auto& [k, v] : delete_bitmap) {
+ charge += v.getSizeInBytes();
+ }
+ return charge;
+}
+
bool DeleteBitmap::contains_agg_without_cache(const BitmapKey& bmk, uint32_t
row_id) const {
std::shared_lock l(lock);
DeleteBitmap::BitmapKey start {std::get<0>(bmk), std::get<1>(bmk), 0};
@@ -1166,6 +1178,52 @@ void DeleteBitmap::merge(const DeleteBitmap& other) {
}
}
+void DeleteBitmap::add_to_remove_queue(
+ const std::string& version_str,
+ const std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey,
DeleteBitmap::BitmapKey>>&
+ vector) {
+ std::shared_lock l(stale_delete_bitmap_lock);
+ _stale_delete_bitmap.emplace(version_str, vector);
+}
+
+void DeleteBitmap::remove_stale_delete_bitmap_from_queue(const
std::vector<std::string>& vector) {
+ std::shared_lock l(stale_delete_bitmap_lock);
+ //<rowset_id, start_version, end_version>
+ std::vector<std::tuple<std::string, uint64_t, uint64_t>> to_delete;
+ auto tablet_id = -1;
+ for (auto& version_str : vector) {
+ auto it = _stale_delete_bitmap.find(version_str);
+ if (it != _stale_delete_bitmap.end()) {
+ auto delete_bitmap_vector = it->second;
+ for (auto& delete_bitmap_tuple : it->second) {
+ if (tablet_id < 0) {
+ tablet_id = std::get<0>(delete_bitmap_tuple);
+ }
+ auto start_bmk = std::get<1>(delete_bitmap_tuple);
+ auto end_bmk = std::get<2>(delete_bitmap_tuple);
+ remove(start_bmk, end_bmk);
+
to_delete.emplace_back(std::make_tuple(std::get<0>(start_bmk).to_string(), 0,
+ std::get<2>(end_bmk)));
+ }
+ _stale_delete_bitmap.erase(version_str);
+ }
+ }
+ if (tablet_id == -1 || to_delete.empty()) {
+ return;
+ }
+ CloudStorageEngine& engine =
ExecEnv::GetInstance()->storage_engine().to_cloud();
+ auto st = engine.meta_mgr().remove_old_version_delete_bitmap(tablet_id,
to_delete);
+ if (!st.ok()) {
+ LOG(WARNING) << "fail to remove_stale_delete_bitmap_from_queue for
tablet=" << tablet_id
+ << ",st=" << st;
+ }
+}
+
+uint64_t DeleteBitmap::get_delete_bitmap_count() {
+ std::shared_lock l(lock);
+ return delete_bitmap.size();
+}
+
// We cannot just copy the underlying memory to construct a string
// due to equivalent objects may have different padding bytes.
// Reading padding bytes is undefined behavior, neither copy nor
diff --git a/be/src/olap/tablet_meta.h b/be/src/olap/tablet_meta.h
index f754f885abe..9017e8baf32 100644
--- a/be/src/olap/tablet_meta.h
+++ b/be/src/olap/tablet_meta.h
@@ -371,6 +371,7 @@ private:
class DeleteBitmap {
public:
mutable std::shared_mutex lock;
+ mutable std::shared_mutex stale_delete_bitmap_lock;
using SegmentId = uint32_t;
using Version = uint64_t;
using BitmapKey = std::tuple<RowsetId, SegmentId, Version>;
@@ -450,6 +451,12 @@ public:
*/
uint64_t cardinality() const;
+ /**
+ * return the total size of the Delete Bitmap(after serialized)
+ */
+
+ size_t get_size() const;
+
/**
* Sets the bitmap of specific segment, it's may be insertion or
replacement
*
@@ -520,6 +527,13 @@ public:
void remove_sentinel_marks();
+ void add_to_remove_queue(const std::string& version_str,
+ const std::vector<std::tuple<int64_t,
DeleteBitmap::BitmapKey,
+
DeleteBitmap::BitmapKey>>& vector);
+ void remove_stale_delete_bitmap_from_queue(const std::vector<std::string>&
vector);
+
+ uint64_t get_delete_bitmap_count();
+
class AggCachePolicy : public LRUCachePolicy {
public:
AggCachePolicy(size_t capacity)
@@ -553,6 +567,10 @@ public:
private:
mutable std::shared_ptr<AggCache> _agg_cache;
int64_t _tablet_id;
+ // <version, <tablet_id, BitmapKeyStart, BitmapKeyEnd>>
+ std::map<std::string,
+ std::vector<std::tuple<int64_t, DeleteBitmap::BitmapKey,
DeleteBitmap::BitmapKey>>>
+ _stale_delete_bitmap;
};
static const std::string SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
diff --git a/be/src/service/http_service.cpp b/be/src/service/http_service.cpp
index 46ee9569992..9330867ded6 100644
--- a/be/src/service/http_service.cpp
+++ b/be/src/service/http_service.cpp
@@ -25,6 +25,7 @@
#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"
@@ -411,6 +412,11 @@ 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));
+ _ev_http_server->register_handler(HttpMethod::GET,
"/api/delete_bitmap/count",
+ count_delete_bitmap_action);
#ifdef ENABLE_INJECTION_POINT
InjectionPointAction* injection_point_action = _pool.add(new
InjectionPointAction);
_ev_http_server->register_handler(HttpMethod::GET,
"/api/injection_point/{op}",
diff --git a/cloud/src/common/bvars.cpp b/cloud/src/common/bvars.cpp
index 21b91416a65..5658a34f3f5 100644
--- a/cloud/src/common/bvars.cpp
+++ b/cloud/src/common/bvars.cpp
@@ -72,6 +72,7 @@ BvarLatencyRecorderWithTag
g_bvar_ms_update_delete_bitmap("ms", "update_delete_b
BvarLatencyRecorderWithTag g_bvar_ms_get_delete_bitmap("ms",
"get_delete_bitmap");
BvarLatencyRecorderWithTag g_bvar_ms_get_delete_bitmap_update_lock("ms",
"get_delete_bitmap_update_lock");
+BvarLatencyRecorderWithTag g_bvar_ms_remove_delete_bitmap("ms",
"remove_delete_bitmap");
BvarLatencyRecorderWithTag g_bvar_ms_get_instance("ms", "get_instance");
BvarLatencyRecorderWithTag g_bvar_ms_get_rl_task_commit_attach("ms",
"get_rl_task_commit_attach");
BvarLatencyRecorderWithTag g_bvar_ms_reset_rl_progress("ms",
"reset_rl_progress");
diff --git a/cloud/src/common/bvars.h b/cloud/src/common/bvars.h
index c5067b053b3..6f368ff06eb 100644
--- a/cloud/src/common/bvars.h
+++ b/cloud/src/common/bvars.h
@@ -171,6 +171,7 @@ extern BvarLatencyRecorderWithTag
g_bvar_ms_finish_tablet_job;
extern BvarLatencyRecorderWithTag g_bvar_ms_update_delete_bitmap;
extern BvarLatencyRecorderWithTag g_bvar_ms_get_delete_bitmap;
extern BvarLatencyRecorderWithTag g_bvar_ms_get_delete_bitmap_update_lock;
+extern BvarLatencyRecorderWithTag g_bvar_ms_remove_delete_bitmap;
extern BvarLatencyRecorderWithTag g_bvar_ms_get_cluster_status;
extern BvarLatencyRecorderWithTag g_bvar_ms_set_cluster_status;
extern BvarLatencyRecorderWithTag g_bvar_ms_get_instance;
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 5f3d64b24c9..17ba2953ce9 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1748,18 +1748,20 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
return;
}
- // 1. Check whether the lock expires
- if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, table_id,
request->lock_id(),
- request->initiator())) {
- LOG(WARNING) << "failed to check delete bitmap lock, table_id=" <<
table_id
- << " request lock_id=" << request->lock_id()
- << " request initiator=" << request->initiator() << "
msg" << msg;
- return;
- }
-
- // 2. Process pending delete bitmap
- if (!process_pending_delete_bitmap(code, msg, ss, txn, instance_id,
tablet_id)) {
- return;
+ bool unlock = request->has_unlock() ? request->unlock() : false;
+ if (!unlock) {
+ // 1. Check whether the lock expires
+ if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id,
table_id, request->lock_id(),
+ request->initiator())) {
+ LOG(WARNING) << "failed to check delete bitmap lock, table_id=" <<
table_id
+ << " request lock_id=" << request->lock_id()
+ << " request initiator=" << request->initiator() << "
msg " << msg;
+ return;
+ }
+ // 2. Process pending delete bitmap
+ if (!process_pending_delete_bitmap(code, msg, ss, txn, instance_id,
tablet_id)) {
+ return;
+ }
}
// 3. store all pending delete bitmap for this txn
@@ -1791,6 +1793,8 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
}
// 4. Update delete bitmap for curent txn
+ size_t total_key = 0;
+ size_t total_size = 0;
for (size_t i = 0; i < request->rowset_ids_size(); ++i) {
auto& key = delete_bitmap_keys.delete_bitmap_keys(i);
auto& val = request->segment_delete_bitmaps(i);
@@ -1814,19 +1818,23 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
msg = "failed to init txn";
return;
}
- if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id,
table_id,
- request->lock_id(),
request->initiator())) {
- LOG(WARNING) << "failed to check delete bitmap lock,
table_id=" << table_id
- << " request lock_id=" << request->lock_id()
- << " request initiator=" << request->initiator()
<< " msg" << msg;
- return;
+ if (!unlock) {
+ if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id,
table_id,
+ request->lock_id(),
request->initiator())) {
+ LOG(WARNING) << "failed to check delete bitmap lock,
table_id=" << table_id
+ << " request lock_id=" << request->lock_id()
+ << " request initiator=" <<
request->initiator() << " msg" << msg;
+ return;
+ }
}
}
// splitting large values (>90*1000) into multiple KVs
cloud::put(txn.get(), key, val, 0);
fdb_txn_size = fdb_txn_size + key.size() + val.size();
- LOG(INFO) << "xxx update delete bitmap put delete_bitmap_key=" <<
hex(key)
- << " lock_id=" << request->lock_id() << " value_size: " <<
val.size();
+ total_key++;
+ total_size += key.size() + val.size();
+ VLOG_DEBUG << "xxx update delete bitmap put delete_bitmap_key=" <<
hex(key)
+ << " lock_id=" << request->lock_id() << " value_size: " <<
val.size();
}
err = txn->commit();
@@ -1836,6 +1844,9 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
msg = ss.str();
return;
}
+ LOG(INFO) << "update_delete_bitmap tablet_id=" << tablet_id << " lock_id="
<< request->lock_id()
+ << " rowset_num=" << request->rowset_ids_size() << " total_key="
<< total_key
+ << " total_size=" << total_size << " unlock=" << unlock;
}
void MetaServiceImpl::get_delete_bitmap(google::protobuf::RpcController*
controller,
@@ -2080,6 +2091,63 @@ void
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcControl
}
}
+void MetaServiceImpl::remove_delete_bitmap(google::protobuf::RpcController*
controller,
+ const RemoveDeleteBitmapRequest*
request,
+ RemoveDeleteBitmapResponse*
response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(remove_delete_bitmap);
+ std::string cloud_unique_id = request->has_cloud_unique_id() ?
request->cloud_unique_id() : "";
+ if (cloud_unique_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "cloud unique id not set";
+ return;
+ }
+
+ instance_id = get_instance_id(resource_mgr_, cloud_unique_id);
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "empty instance_id";
+ LOG(WARNING) << msg << ", cloud_unique_id=" <<
request->cloud_unique_id();
+ return;
+ }
+ RPC_RATE_LIMIT(remove_delete_bitmap)
+ auto tablet_id = request->tablet_id();
+ auto& rowset_ids = request->rowset_ids();
+ auto& begin_versions = request->begin_versions();
+ auto& end_versions = request->end_versions();
+ if (rowset_ids.size() != begin_versions.size() || rowset_ids.size() !=
end_versions.size()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ ss << "rowset and version size not match. "
+ << " rowset_size=" << rowset_ids.size()
+ << " begin_version_size=" << begin_versions.size()
+ << " end_version_size=" << end_versions.size();
+ msg = ss.str();
+ return;
+ }
+ std::unique_ptr<Transaction> txn;
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ LOG(WARNING) << "failed to init txn";
+ return;
+ }
+ for (size_t i = 0; i < rowset_ids.size(); i++) {
+ auto delete_bitmap_start = meta_delete_bitmap_key(
+ {instance_id, tablet_id, rowset_ids[i], begin_versions[i], 0});
+ auto delete_bitmap_end = meta_delete_bitmap_key(
+ {instance_id, tablet_id, rowset_ids[i], end_versions[i],
INT64_MAX});
+ txn->remove(delete_bitmap_start, delete_bitmap_end);
+ }
+ err = txn->commit();
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::COMMIT>(err);
+ ss << "failed to commit job kv, err=" << err;
+ msg = ss.str();
+ return;
+ }
+ LOG(INFO) << "remove_delete_bitmap,tablet_id=" << tablet_id
+ << ",rowset_num=" << rowset_ids.size();
+}
+
std::pair<MetaServiceCode, std::string> MetaServiceImpl::get_instance_info(
const std::string& instance_id, const std::string& cloud_unique_id,
InstanceInfoPB* instance) {
diff --git a/cloud/src/meta-service/meta_service.h
b/cloud/src/meta-service/meta_service.h
index 5415e7ce1e6..a632d48efcc 100644
--- a/cloud/src/meta-service/meta_service.h
+++ b/cloud/src/meta-service/meta_service.h
@@ -265,6 +265,11 @@ public:
GetDeleteBitmapUpdateLockResponse*
response,
::google::protobuf::Closure* done)
override;
+ void remove_delete_bitmap(google::protobuf::RpcController* controller,
+ const RemoveDeleteBitmapRequest* request,
+ RemoveDeleteBitmapResponse* response,
+ ::google::protobuf::Closure* done) override;
+
// cloud control get cluster's status by this api
void get_cluster_status(google::protobuf::RpcController* controller,
const GetClusterStatusRequest* request,
@@ -631,6 +636,13 @@ public:
done);
}
+ void remove_delete_bitmap(google::protobuf::RpcController* controller,
+ const RemoveDeleteBitmapRequest* request,
+ RemoveDeleteBitmapResponse* response,
+ ::google::protobuf::Closure* done) override {
+ call_impl(&cloud::MetaService::remove_delete_bitmap, controller,
request, response, done);
+ }
+
// cloud control get cluster's status by this api
void get_cluster_status(google::protobuf::RpcController* controller,
const GetClusterStatusRequest* request,
diff --git a/cloud/src/meta-service/meta_service_helper.h
b/cloud/src/meta-service/meta_service_helper.h
index 664d400b236..bb1f6197f1d 100644
--- a/cloud/src/meta-service/meta_service_helper.h
+++ b/cloud/src/meta-service/meta_service_helper.h
@@ -58,6 +58,9 @@ void begin_rpc(std::string_view func_name, brpc::Controller*
ctrl, const Request
std::is_same_v<Request, GetTabletRequest>) {
VLOG_DEBUG << "begin " << func_name << " from " << ctrl->remote_side()
<< " request=" << req->ShortDebugString();
+ } else if constexpr (std::is_same_v<Request, RemoveDeleteBitmapRequest>) {
+ LOG(INFO) << "begin " << func_name << " from " << ctrl->remote_side()
+ << " tablet_id=" << req->tablet_id() << " rowset_size=" <<
req->rowset_ids_size();
} else {
LOG(INFO) << "begin " << func_name << " from " << ctrl->remote_side()
<< " request=" << req->ShortDebugString();
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index 268744a0088..93420bddbf6 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -1384,6 +1384,7 @@ message UpdateDeleteBitmapRequest {
repeated int64 versions = 9;
// Serialized roaring bitmaps indexed with {rowset_id, segment_id, version}
repeated bytes segment_delete_bitmaps = 10;
+ optional bool unlock = 11;
}
message UpdateDeleteBitmapResponse {
@@ -1411,6 +1412,18 @@ message GetDeleteBitmapResponse {
repeated bytes segment_delete_bitmaps = 5;
}
+message RemoveDeleteBitmapRequest {
+ optional string cloud_unique_id = 1; // For auth
+ optional int64 tablet_id = 2;
+ repeated string rowset_ids = 3;
+ repeated int64 begin_versions = 4;
+ repeated int64 end_versions = 5;
+}
+
+message RemoveDeleteBitmapResponse {
+ optional MetaServiceResponseStatus status = 1;
+}
+
message PendingDeleteBitmapPB {
repeated bytes delete_bitmap_keys = 1;
}
@@ -1550,6 +1563,7 @@ service MetaService {
rpc update_delete_bitmap(UpdateDeleteBitmapRequest)
returns(UpdateDeleteBitmapResponse);
rpc get_delete_bitmap(GetDeleteBitmapRequest)
returns(GetDeleteBitmapResponse);
rpc get_delete_bitmap_update_lock(GetDeleteBitmapUpdateLockRequest)
returns(GetDeleteBitmapUpdateLockResponse);
+ rpc remove_delete_bitmap(RemoveDeleteBitmapRequest)
returns(RemoveDeleteBitmapResponse);
// routine load progress
rpc get_rl_task_commit_attach(GetRLTaskCommitAttachRequest) returns
(GetRLTaskCommitAttachResponse);
diff --git
a/regression-test/data/compaction/test_cu_compaction_remove_old_version_delete_bitmap.out
b/regression-test/data/compaction/test_cu_compaction_remove_old_version_delete_bitmap.out
new file mode 100644
index 00000000000..1c3611fe0b7
--- /dev/null
+++
b/regression-test/data/compaction/test_cu_compaction_remove_old_version_delete_bitmap.out
@@ -0,0 +1,29 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !sql --
+0 0 0
+1 8 8
+
+-- !sql --
+0 0 0
+1 8 8
+
+-- !sql --
+0 0 0
+1 13 13
+
+-- !sql --
+0 0 0
+1 13 13
+
+-- !sql --
+0 0 0
+1 23 23
+
+-- !sql --
+0 0 0
+1 23 23
+
+-- !sql --
+0 0 0
+1 28 28
+
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
new file mode 100644
index 00000000000..2219cc175b5
--- /dev/null
+++
b/regression-test/suites/compaction/test_cu_compaction_remove_old_version_delete_bitmap.groovy
@@ -0,0 +1,316 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite("test_cu_compaction_remove_old_version_delete_bitmap", "nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+ def backendId_to_backendIP = [:]
+ def backendId_to_backendHttpPort = [:]
+ def backendId_to_params = [string: [:]]
+ getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+ def set_be_param = { paramName, paramValue ->
+ // for eache be node, set paramName=paramValue
+ for (String id in backendId_to_backendIP.keySet()) {
+ def beIp = backendId_to_backendIP.get(id)
+ def bePort = backendId_to_backendHttpPort.get(id)
+ def (code, out, err) = curl("POST",
String.format("http://%s:%s/api/update_config?%s=%s";, beIp, bePort, paramName,
paramValue))
+ assertTrue(out.contains("OK"))
+ }
+ }
+
+ def reset_be_param = { paramName ->
+ // for eache be node, reset paramName to default
+ for (String id in backendId_to_backendIP.keySet()) {
+ def beIp = backendId_to_backendIP.get(id)
+ def bePort = backendId_to_backendHttpPort.get(id)
+ def original_value = backendId_to_params.get(id).get(paramName)
+ def (code, out, err) = curl("POST",
String.format("http://%s:%s/api/update_config?%s=%s";, beIp, bePort, paramName,
original_value))
+ assertTrue(out.contains("OK"))
+ }
+ }
+
+ def get_be_param = { paramName ->
+ // for eache be node, get param value by default
+ def paramValue = ""
+ for (String id in backendId_to_backendIP.keySet()) {
+ def beIp = backendId_to_backendIP.get(id)
+ def bePort = backendId_to_backendHttpPort.get(id)
+ // get the config value from be
+ def (code, out, err) = curl("GET",
String.format("http://%s:%s/api/show_config?conf_item=%s";, beIp, bePort,
paramName))
+ assertTrue(code == 0)
+ assertTrue(out.contains(paramName))
+ // parsing
+ def resultList = parseJson(out)[0]
+ assertTrue(resultList.size() == 4)
+ // get original value
+ paramValue = resultList[2]
+ backendId_to_params.get(id, [:]).put(paramName, paramValue)
+ }
+ }
+
+ def triggerCompaction = { be_host, be_http_port, compact_type, tablet_id ->
+ if (compact_type == "cumulative") {
+ def (code_1, out_1, err_1) = be_run_cumulative_compaction(be_host,
be_http_port, tablet_id)
+ logger.info("Run compaction: code=" + code_1 + ", out=" + out_1 +
", err=" + err_1)
+ assertEquals(code_1, 0)
+ return out_1
+ } else if (compact_type == "full") {
+ def (code_2, out_2, err_2) = be_run_full_compaction(be_host,
be_http_port, tablet_id)
+ logger.info("Run compaction: code=" + code_2 + ", out=" + out_2 +
", err=" + err_2)
+ assertEquals(code_2, 0)
+ return out_2
+ } else {
+ assertFalse(True)
+ }
+ }
+
+ def getTabletStatus = { be_host, be_http_port, tablet_id ->
+ boolean running = true
+ Thread.sleep(1000)
+ StringBuilder sb = new StringBuilder();
+ sb.append("curl -X GET http://${be_host}:${be_http_port}";)
+ sb.append("/api/compaction/show?tablet_id=")
+ sb.append(tablet_id)
+
+ String command = sb.toString()
+ logger.info(command)
+ process = command.execute()
+ code = process.waitFor()
+ out = process.getText()
+ logger.info("Get tablet status: =" + code + ", out=" + out)
+ assertEquals(code, 0)
+ def tabletStatus = parseJson(out.trim())
+ return tabletStatus
+ }
+
+ def waitForCompaction = { be_host, be_http_port, tablet_id ->
+ boolean running = true
+ do {
+ Thread.sleep(1000)
+ StringBuilder sb = new StringBuilder();
+ sb.append("curl -X GET http://${be_host}:${be_http_port}";)
+ sb.append("/api/compaction/run_status?tablet_id=")
+ sb.append(tablet_id)
+
+ String command = sb.toString()
+ logger.info(command)
+ process = command.execute()
+ code = process.waitFor()
+ out = process.getText()
+ logger.info("Get compaction status: code=" + code + ", out=" + out)
+ assertEquals(code, 0)
+ def compactionStatus = parseJson(out.trim())
+ assertEquals("success", compactionStatus.status.toLowerCase())
+ running = compactionStatus.run_status
+ } while (running)
+ }
+
+ def getDeleteBitmapStatus = { be_host, be_http_port, tablet_id ->
+ boolean running = true
+ StringBuilder sb = new StringBuilder();
+ sb.append("curl -X GET http://${be_host}:${be_http_port}";)
+ sb.append("/api/delete_bitmap/count?tablet_id=")
+ sb.append(tablet_id)
+
+ String command = sb.toString()
+ logger.info(command)
+ process = command.execute()
+ code = process.waitFor()
+ out = process.getText()
+ logger.info("Get delete bitmap count status: =" + code + ", out=" +
out)
+ assertEquals(code, 0)
+ def deleteBitmapStatus = parseJson(out.trim())
+ return deleteBitmapStatus
+ }
+
+ def testTable = "test_cu_compaction_remove_old_version_delete_bitmap"
+ def timeout = 10000
+ sql """ DROP TABLE IF EXISTS ${testTable}"""
+ def testTableDDL = """
+ create table ${testTable}
+ (
+ `plan_id` bigint(20) NOT NULL,
+ `target_id` int(20) NOT NULL,
+ `target_name` varchar(255) NOT NULL
+ )
+ ENGINE=OLAP
+ UNIQUE KEY(`plan_id`)
+ COMMENT 'OLAP'
+ DISTRIBUTED BY HASH(`plan_id`) BUCKETS 1
+ PROPERTIES (
+ "enable_unique_key_merge_on_write" = "true",
+ "replication_allocation" = "tag.location.default: 1",
+ "disable_auto_compaction" = "true"
+ );
+ """
+ sql testTableDDL
+ sql "sync"
+
+ // store the original value
+ get_be_param("compaction_promotion_version_count")
+ get_be_param("tablet_rowset_stale_sweep_time_sec")
+ set_be_param("compaction_promotion_version_count", "5")
+ set_be_param("tablet_rowset_stale_sweep_time_sec", "0")
+
+ try {
+
GetDebugPoint().enableDebugPointForAllBEs("CloudCumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets")
+ // 1. test normal
+ sql "sync"
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,1,'1'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,2,'2'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,3,'3'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,4,'4'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,5,'5'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,6,'6'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,7,'7'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,8,'8'); """
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
+ // trigger compaction to generate base rowset
+ def tablets = sql_return_maparray """ show tablets from ${testTable};
"""
+ logger.info("tablets: " + tablets)
+ def delete_bitmap_count = 0
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ def tablet_info = sql_return_maparray """ show tablet
${tablet_id}; """
+ logger.info("tablet: " + tablet_info)
+ String trigger_backend_id = tablet.BackendId
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+
+ // before compaction, delete_bitmap_count is (rowsets num - 1)
+ delete_bitmap_count =
getDeleteBitmapStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id).delete_bitmap_count
+ assertTrue(delete_bitmap_count == 7)
+ logger.info("delete_bitmap_count:" + delete_bitmap_count)
+
+
assertTrue(triggerCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id],
+ "cumulative", tablet_id).contains("Success"));
+ waitForCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id)
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+ }
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
+ def now = System.currentTimeMillis()
+
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,9,'9'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,10,'10'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,11,'11'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,12,'12'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,13,'13'); """
+
+ def time_diff = System.currentTimeMillis() - now
+ logger.info("time_diff:" + time_diff)
+ assertTrue(time_diff <= timeout, "wait_for_insert_into_values timeout")
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
+ // trigger cu compaction to remove old version delete bitmap
+
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ def tablet_info = sql_return_maparray """ show tablet
${tablet_id}; """
+ logger.info("tablet: " + tablet_info)
+
+ // before compaction, delete_bitmap_count is (rowsets num - 1)
+ String trigger_backend_id = tablet.BackendId
+ delete_bitmap_count =
getDeleteBitmapStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id).delete_bitmap_count
+ logger.info("delete_bitmap_count:" + delete_bitmap_count)
+ assertTrue(delete_bitmap_count == 12)
+
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+
assertTrue(triggerCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id],
+ "cumulative", tablet_id).contains("Success"));
+ waitForCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id)
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+
+ Thread.sleep(1000)
+ // after compaction, delete_bitmap_count is 1
+ delete_bitmap_count =
getDeleteBitmapStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id).delete_bitmap_count
+ logger.info("delete_bitmap_count:" + delete_bitmap_count)
+ assertTrue(delete_bitmap_count == 1)
+ }
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
+ // 2. test update delete bitmap failed
+
+ now = System.currentTimeMillis()
+
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,19,'19'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,20,'20'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,21,'21'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,22,'22'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,23,'23'); """
+
+ time_diff = System.currentTimeMillis() - now
+ logger.info("time_diff:" + time_diff)
+ assertTrue(time_diff <= timeout, "wait_for_insert_into_values timeout")
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
GetDebugPoint().enableDebugPointForAllBEs("CloudCumulativeCompaction.modify_rowsets.update_delete_bitmap_failed")
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ def tablet_info = sql_return_maparray """ show tablet
${tablet_id}; """
+ logger.info("tablet: " + tablet_info)
+ String trigger_backend_id = tablet.BackendId
+
+ delete_bitmap_count =
getDeleteBitmapStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id).delete_bitmap_count
+ assertTrue(delete_bitmap_count == 6)
+ logger.info("delete_bitmap_count:" + delete_bitmap_count)
+
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+
assertTrue(triggerCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id],
+ "cumulative", tablet_id).contains("Success"));
+ waitForCompaction(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id)
+ getTabletStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id);
+
+ // update fail, delete_bitmap_count will not change
+ Thread.sleep(1000)
+ delete_bitmap_count =
getDeleteBitmapStatus(backendId_to_backendIP[trigger_backend_id],
backendId_to_backendHttpPort[trigger_backend_id], tablet_id).delete_bitmap_count
+ assertTrue(delete_bitmap_count == 6)
+ logger.info("delete_bitmap_count:" + delete_bitmap_count)
+ }
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
+ now = System.currentTimeMillis()
+
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,24,'24'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,25,'25'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,26,'26'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,27,'27'); """
+ sql """ INSERT INTO ${testTable} VALUES (0,0,'0'),(1,28,'28'); """
+
+ time_diff = System.currentTimeMillis() - now
+ logger.info("time_diff:" + time_diff)
+ assertTrue(time_diff <= timeout, "wait_for_insert_into_values timeout")
+
+ qt_sql "select * from ${testTable} order by plan_id"
+
GetDebugPoint().disableDebugPointForAllBEs("CloudCumulativeCompaction.modify_rowsets.update_delete_bitmap_failed")
+ } finally {
+ reset_be_param("compaction_promotion_version_count")
+ reset_be_param("tablet_rowset_stale_sweep_time_sec")
+
GetDebugPoint().disableDebugPointForAllBEs("CloudCumulativeCompaction.modify_rowsets.delete_expired_stale_rowsets")
+
GetDebugPoint().disableDebugPointForAllBEs("CloudCumulativeCompaction.modify_rowsets.update_delete_bitmap_failed")
+ }
+
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
