Merge branch 'master' into ttl-management-rebased

Author: freemandealer

be/src/agent/cgroup_cpu_ctl.cpp

@@ -292,7 +292,7 @@ Status CgroupCpuCtl::delete_unused_cgroup_path(std::set<uint64_t>& used_wg_ids)
         std::string wg_path = query_path + unused_wg_id;
         int ret = rmdir(wg_path.c_str());
         if (ret < 0) {
-            LOG(WARNING) << "rmdir failed, path=" << wg_path;
+            LOG(WARNING) << "remove cgroup path failed, path=" << wg_path << ", error=" << ret;
             failed_count++;
         }
     }
@@ -317,7 +317,8 @@ Status CgroupV1CpuCtl::init() {
     if (access(_cgroup_v1_cpu_tg_path.c_str(), F_OK) != 0) {
         int ret = mkdir(_cgroup_v1_cpu_tg_path.c_str(), S_IRWXU);
         if (ret != 0) {
-            LOG(ERROR) << "cgroup v1 mkdir workload group failed, path=" << _cgroup_v1_cpu_tg_path;
+            LOG(WARNING) << "cgroup v1 make workload group dir failed, path="
+                         << _cgroup_v1_cpu_tg_path << ", error=" << ret;
             return Status::InternalError<false>("cgroup v1 mkdir workload group failed, path={}",
                                                 _cgroup_v1_cpu_tg_path);
         }
@@ -382,6 +383,8 @@ Status CgroupV2CpuCtl::init() {
     if (access(_cgroup_v2_query_wg_path.c_str(), F_OK) != 0) {
         int ret = mkdir(_cgroup_v2_query_wg_path.c_str(), S_IRWXU);
         if (ret != 0) {
+            LOG(WARNING) << "cgroup v2 make workload group dir failed, path="
+                         << _cgroup_v2_query_wg_path << ", error=" << ret;
             return Status::InternalError<false>("cgroup v2 mkdir wg failed, path={}",
                                                 _cgroup_v2_query_wg_path);
         }

be/src/agent/task_worker_pool.cpp

@@ -419,6 +419,8 @@ Status _submit_task(const TAgentTaskRequest& task,
     // TODO(plat1ko): check task request member
 
     // Set the receiving time of task so that we can determine whether it is timed out later
+    // exist a path task_worker_pool <- agent_server <- backend_service <- BackendService
+    // use the arg BackendService_submit_tasks_args.tasks is not const, so modify is ok
     (const_cast<TAgentTaskRequest&>(task)).__set_recv_time(time(nullptr));
     auto st = submit_op(task);
     if (!st.ok()) [[unlikely]] {
@@ -617,7 +619,7 @@ Status PriorTaskWorkerPool::submit_task(const TAgentTaskRequest& task) {
     });
 }
 
-Status PriorTaskWorkerPool::submit_high_prior_and_cancel_low(const TAgentTaskRequest& task) {
+Status PriorTaskWorkerPool::submit_high_prior_and_cancel_low(TAgentTaskRequest& task) {
     const TTaskType::type task_type = task.task_type;
     int64_t signature = task.signature;
     std::string type_str;
@@ -657,7 +659,7 @@ Status PriorTaskWorkerPool::submit_high_prior_and_cancel_low(const TAgentTaskReq
     } while (false);
 
     // Set the receiving time of task so that we can determine whether it is timed out later
-    (const_cast<TAgentTaskRequest&>(task)).__set_recv_time(time(nullptr));
+    task.__set_recv_time(time(nullptr));
 
     LOG_INFO("successfully submit task").tag("type", type_str).tag("signature", signature);
     return Status::OK();
@@ -1926,6 +1928,9 @@ void push_callback(StorageEngine& engine, const TAgentTaskRequest& req) {
               << " push_type=" << push_req.push_type;
     std::vector<TTabletInfo> tablet_infos;
 
+    // exist a path task_worker_pool <- agent_server <- backend_service <- BackendService
+    // use the arg BackendService_submit_tasks_args.tasks is not const
+    // and push_req will be modify, so modify is ok
     EngineBatchLoadTask engine_task(engine, const_cast<TPushReq&>(push_req), &tablet_infos);
     SCOPED_ATTACH_TASK(engine_task.mem_tracker());
     auto status = engine_task.execute();

be/src/agent/task_worker_pool.h

@@ -88,7 +88,7 @@ class PriorTaskWorkerPool final : public TaskWorkerPoolIf {
 
     Status submit_task(const TAgentTaskRequest& task) override;
 
-    Status submit_high_prior_and_cancel_low(const TAgentTaskRequest& task);
+    Status submit_high_prior_and_cancel_low(TAgentTaskRequest& task);
 
 private:
     void normal_loop();

be/src/cloud/cloud_backend_service.cpp

@@ -32,6 +32,7 @@
 #include "runtime/stream_load/stream_load_context.h"
 #include "runtime/stream_load/stream_load_recorder.h"
 #include "util/brpc_client_cache.h" // BrpcClientCache
+#include "util/stack_util.h"
 #include "util/thrift_server.h"
 
 namespace doris {
@@ -213,8 +214,12 @@ void CloudBackendService::warm_up_cache_async(TWarmUpCacheAsyncResponse& respons
     }
     brpc::Controller cntl;
     PGetFileCacheMetaRequest brpc_request;
-    std::for_each(request.tablet_ids.cbegin(), request.tablet_ids.cend(),
-                  [&](int64_t tablet_id) { brpc_request.add_tablet_ids(tablet_id); });
+    std::stringstream ss;
+    std::for_each(request.tablet_ids.cbegin(), request.tablet_ids.cend(), [&](int64_t tablet_id) {
+        brpc_request.add_tablet_ids(tablet_id);
+        ss << tablet_id << ",";
+    });
+    VLOG_DEBUG << "tablets set: " << ss.str() << " stack: " << get_stack_trace();
     PGetFileCacheMetaResponse brpc_response;
 
     brpc_stub->get_file_cache_meta_by_tablet_id(&cntl, &brpc_request, &brpc_response, nullptr);

be/src/cloud/cloud_meta_mgr.cpp

@@ -388,6 +388,7 @@ Status retry_rpc(std::string_view op_name, const Request& req, Response* res,
     static_assert(std::is_base_of_v<::google::protobuf::Message, Request>);
     static_assert(std::is_base_of_v<::google::protobuf::Message, Response>);
 
+    // Applies only to the current file, and all req are non-const, but passed as const types.
     const_cast<Request&>(req).set_request_ip(BackendOptions::get_be_endpoint());
 
     int retry_times = 0;
@@ -944,6 +945,17 @@ Status CloudMetaMgr::sync_tablet_delete_bitmap(CloudTablet* tablet, int64_t old_
         DeleteBitmapPtr new_delete_bitmap = std::make_shared<DeleteBitmap>(tablet->tablet_id());
         *delete_bitmap = *new_delete_bitmap;
     }
+
+    if (read_version == 2 && config::delete_bitmap_store_write_version == 1) {
+        return Status::InternalError(
+                "please set delete_bitmap_store_read_version to 1 or 3 because "
+                "delete_bitmap_store_write_version is 1");
+    } else if (read_version == 1 && config::delete_bitmap_store_write_version == 2) {
+        return Status::InternalError(
+                "please set delete_bitmap_store_read_version to 2 or 3 because "
+                "delete_bitmap_store_write_version is 2");
+    }
+
     int64_t new_max_version = std::max(old_max_version, rs_metas.rbegin()->end_version());
     // When there are many delete bitmaps that need to be synchronized, it
     // may take a longer time, especially when loading the tablet for the
@@ -1637,6 +1649,7 @@ Status CloudMetaMgr::update_delete_bitmap(const CloudTablet& tablet, int64_t loc
     if (next_visible_version > 0) {
         req.set_next_visible_version(next_visible_version);
     }
+    req.set_store_version(store_version);
 
     bool write_v1 = store_version == 1 || store_version == 3;
     bool write_v2 = store_version == 2 || store_version == 3;
@@ -1784,6 +1797,10 @@ Status CloudMetaMgr::cloud_update_delete_bitmap_without_lock(
         const CloudTablet& tablet, DeleteBitmap* delete_bitmap,
         std::map<std::string, int64_t>& rowset_to_versions, int64_t pre_rowset_agg_start_version,
         int64_t pre_rowset_agg_end_version) {
+    if (config::delete_bitmap_store_write_version == 2) {
+        VLOG_DEBUG << "no need to agg delete bitmap v1 in ms because use v2";
+        return Status::OK();
+    }
     LOG(INFO) << "cloud_update_delete_bitmap_without_lock, tablet_id: " << tablet.tablet_id()
               << ", delete_bitmap size: " << delete_bitmap->delete_bitmap.size();
     UpdateDeleteBitmapRequest req;
@@ -1915,12 +1932,12 @@ void CloudMetaMgr::remove_delete_bitmap_update_lock(int64_t table_id, int64_t lo
     }
 }
 
-void CloudMetaMgr::check_table_size_correctness(const RowsetMeta& rs_meta) {
+void CloudMetaMgr::check_table_size_correctness(RowsetMeta& rs_meta) {
     if (!config::enable_table_size_correctness_check) {
         return;
     }
     int64_t total_segment_size = get_segment_file_size(rs_meta);
-    int64_t total_inverted_index_size = get_inverted_index_file_szie(rs_meta);
+    int64_t total_inverted_index_size = get_inverted_index_file_size(rs_meta);
     if (rs_meta.data_disk_size() != total_segment_size ||
         rs_meta.index_disk_size() != total_inverted_index_size ||
         rs_meta.data_disk_size() + rs_meta.index_disk_size() != rs_meta.total_disk_size()) {
@@ -1939,9 +1956,9 @@ void CloudMetaMgr::check_table_size_correctness(const RowsetMeta& rs_meta) {
     }
 }
 
-int64_t CloudMetaMgr::get_segment_file_size(const RowsetMeta& rs_meta) {
+int64_t CloudMetaMgr::get_segment_file_size(RowsetMeta& rs_meta) {
     int64_t total_segment_size = 0;
-    const auto fs = const_cast<RowsetMeta&>(rs_meta).fs();
+    const auto fs = rs_meta.fs();
     if (!fs) {
         LOG(WARNING) << "get fs failed, resource_id={}" << rs_meta.resource_id();
     }
@@ -1966,9 +1983,9 @@ int64_t CloudMetaMgr::get_segment_file_size(const RowsetMeta& rs_meta) {
     return total_segment_size;
 }
 
-int64_t CloudMetaMgr::get_inverted_index_file_szie(const RowsetMeta& rs_meta) {
+int64_t CloudMetaMgr::get_inverted_index_file_size(RowsetMeta& rs_meta) {
     int64_t total_inverted_index_size = 0;
-    const auto fs = const_cast<RowsetMeta&>(rs_meta).fs();
+    const auto fs = rs_meta.fs();
     if (!fs) {
         LOG(WARNING) << "get fs failed, resource_id={}" << rs_meta.resource_id();
     }

be/src/cloud/cloud_meta_mgr.h

@@ -191,9 +191,9 @@ class CloudMetaMgr {
                                                GetDeleteBitmapResponse& res,
                                                int64_t bytes_threadhold);
 
-    void check_table_size_correctness(const RowsetMeta& rs_meta);
-    int64_t get_segment_file_size(const RowsetMeta& rs_meta);
-    int64_t get_inverted_index_file_szie(const RowsetMeta& rs_meta);
+    void check_table_size_correctness(RowsetMeta& rs_meta);
+    int64_t get_segment_file_size(RowsetMeta& rs_meta);
+    int64_t get_inverted_index_file_size(RowsetMeta& rs_meta);
 };
 
 } // namespace cloud

be/src/cloud/cloud_tablet.cpp

@@ -61,6 +61,7 @@
 #include "olap/tablet_schema.h"
 #include "olap/txn_manager.h"
 #include "util/debug_points.h"
+#include "util/stack_util.h"
 #include "vec/common/schema_util.h"
 
 namespace doris {
@@ -382,6 +383,8 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_
         return;
     }
 
+    VLOG_DEBUG << "add_rowsets tablet_id=" << tablet_id() << " stack: " << get_stack_trace();
+
     auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) {
         for (auto& rs : rowsets) {
             if (version_overlap || warmup_delta_data) {

be/src/cloud/cloud_tablet_mgr.cpp

@@ -29,6 +29,7 @@
 #include "common/status.h"
 #include "olap/lru_cache.h"
 #include "runtime/memory/cache_policy.h"
+#include "util/stack_util.h"
 
 namespace doris {
 uint64_t g_tablet_report_inactive_duration_ms = 0;
@@ -175,6 +176,8 @@ Result<std::shared_ptr<CloudTablet>> CloudTabletMgr::get_tablet(int64_t tablet_i
         TabletMap& tablet_map;
     };
 
+    VLOG_DEBUG << "get_tablet tablet_id=" << tablet_id << " stack: " << get_stack_trace();
+
     auto tablet_id_str = std::to_string(tablet_id);
     CacheKey key(tablet_id_str);
     auto* handle = _cache->lookup(key);

be/src/cloud/cloud_warm_up_manager.cpp

@@ -39,6 +39,7 @@
 #include "runtime/client_cache.h"
 #include "runtime/exec_env.h"
 #include "util/brpc_client_cache.h" // BrpcClientCache
+#include "util/stack_util.h"
 #include "util/thrift_rpc_helper.h"
 #include "util/time.h"
 
@@ -210,6 +211,7 @@ void CloudWarmUpManager::handle_jobs() {
                 std::make_shared<bthread::CountdownEvent>(0);
 
         for (int64_t tablet_id : cur_job->tablet_ids) {
+            VLOG_DEBUG << "Warm up tablet " << tablet_id << " stack: " << get_stack_trace();
             if (_cur_job_id == 0) { // The job is canceled
                 break;
             }

be/src/common/config.cpp

@@ -317,7 +317,7 @@ DEFINE_Validator(task_executor_initial_max_concurrency_per_task, [](const int co
     return true;
 });
 // Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "false");
+DEFINE_Bool(enable_task_executor_in_internal_table, "true");
 // Enable task executor in external table scan.
 DEFINE_Bool(enable_task_executor_in_external_table, "true");