Merge branch 'master' into var-pred-be

.asf.yaml

@@ -49,6 +49,7 @@ github:
         strict: false
         contexts:
           - License Check
+          - Clang Formatter
           - CheckStyle
           - P0 Regression (Doris Regression)
           - External Regression (Doris External Regression)
@@ -86,6 +87,7 @@ github:
         strict: false
         contexts:
           - License Check
+          - Clang Formatter
           - CheckStyle
           - Build Broker
           - ShellCheck
@@ -107,6 +109,7 @@ github:
         strict: false
         contexts:
           - License Check
+          - Clang Formatter
           - CheckStyle
           - P0 Regression (Doris Regression)
           - External Regression (Doris External Regression)

.github/CODEOWNERS

@@ -15,5 +15,6 @@
 # limitations under the License.
 #
 be/src/io/* @platoneko @gavinchou @dataroaring
+be/src/agent/be_exec_version_manager.cpp @BiteTheDDDDt
 fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @dataroaring @CalvinKirs @morningman
 **/pom.xml @CalvinKirs @morningman

.github/workflows/clang-format.yml

@@ -64,7 +64,7 @@ jobs:
           git clone https://github.com/DoozyX/clang-format-lint-action .github/actions/clang-format-lint-action
 
           pushd .github/actions/clang-format-lint-action &>/dev/null
-          git checkout 6adbe14579e5b8e19eb3e31e5ff2479f3bd302c7
+          git checkout c71d0bf4e21876ebec3e5647491186f8797fde31 # v0.18.2
           popd &>/dev/null
 
       - name: Install Python dependencies

be/src/agent/task_worker_pool.cpp

@@ -102,7 +102,7 @@ namespace {
 std::mutex s_task_signatures_mtx;
 std::unordered_map<TTaskType::type, std::unordered_set<int64_t>> s_task_signatures;
 
-std::atomic_ulong s_report_version(time(nullptr) * 10000);
+std::atomic_ulong s_report_version(time(nullptr) * 100000);
 
 void increase_report_version() {
     s_report_version.fetch_add(1, std::memory_order_relaxed);
@@ -1074,6 +1074,7 @@ void report_tablet_callback(StorageEngine& engine, const TMasterInfo& master_inf
     request.__set_backend(BackendOptions::get_local_backend());
     request.__isset.tablets = true;
 
+    increase_report_version();
     uint64_t report_version;
     for (int i = 0; i < 5; i++) {
         request.tablets.clear();

be/src/cloud/cloud_meta_mgr.cpp

@@ -906,7 +906,7 @@ Status CloudMetaMgr::precommit_txn(const StreamLoadContext& ctx) {
     return retry_rpc("precommit txn", req, &res, &MetaService_Stub::precommit_txn);
 }
 
-Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos) {
+Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos, bool* is_vault_mode) {
     GetObjStoreInfoRequest req;
     GetObjStoreInfoResponse resp;
     req.set_cloud_unique_id(config::cloud_unique_id);
@@ -916,6 +916,8 @@ Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos) {
         return s;
     }
 
+    *is_vault_mode = resp.enable_storage_vault();
+
     auto add_obj_store = [&vault_infos](const auto& obj_store) {
         vault_infos->emplace_back(obj_store.id(), S3Conf::get_s3_conf(obj_store),
                                   StorageVaultPB_PathFormat {});
@@ -931,6 +933,7 @@ Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos) {
         }
     });
 
+    // desensitization, hide secret
     for (int i = 0; i < resp.obj_info_size(); ++i) {
         resp.mutable_obj_info(i)->set_sk(resp.obj_info(i).sk().substr(0, 2) + "xxx");
     }
@@ -940,7 +943,8 @@ Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos) {
         j->mutable_obj_info()->set_sk(j->obj_info().sk().substr(0, 2) + "xxx");
     }
 
-    LOG(INFO) << "get storage vault response: " << resp.ShortDebugString();
+    LOG(INFO) << "get storage vault, enable_storage_vault=" << is_vault_mode
+              << " response=" << resp.ShortDebugString();
     return Status::OK();
 }
 

be/src/cloud/cloud_meta_mgr.h

@@ -73,7 +73,14 @@ class CloudMetaMgr {
 
     Status precommit_txn(const StreamLoadContext& ctx);
 
-    Status get_storage_vault_info(StorageVaultInfos* vault_infos);
+    /**
+     * Gets storage vault (storage backends) from meta-service
+     * 
+     * @param vault_info output param, all storage backends
+     * @param is_vault_mode output param, true for pure vault mode, false for legacy mode
+     * @return status
+     */
+    Status get_storage_vault_info(StorageVaultInfos* vault_infos, bool* is_vault_mode);
 
     Status prepare_tablet_job(const TabletJobInfoPB& job, StartTabletJobResponse* res);
 

be/src/cloud/cloud_storage_engine.cpp

@@ -161,8 +161,9 @@ struct RefreshFSVaultVisitor {
 
 Status CloudStorageEngine::open() {
     cloud::StorageVaultInfos vault_infos;
+    bool enable_storage_vault = false;
     do {
-        auto st = _meta_mgr->get_storage_vault_info(&vault_infos);
+        auto st = _meta_mgr->get_storage_vault_info(&vault_infos, &enable_storage_vault);
         if (st.ok()) {
             break;
         }
@@ -177,7 +178,11 @@ Status CloudStorageEngine::open() {
             return vault_process_error(id, vault_info, std::move(st));
         }
     }
-    set_latest_fs(get_filesystem(std::get<0>(vault_infos.back())));
+
+    // vault mode should not support latest_fs to get rid of unexpected storage backends choosen
+    if (!enable_storage_vault) {
+        set_latest_fs(get_filesystem(std::get<0>(vault_infos.back())));
+    }
 
     // TODO(plat1ko): DeleteBitmapTxnManager
 
@@ -340,7 +345,8 @@ void CloudStorageEngine::_check_file_cache_ttl_block_valid() {
 
 void CloudStorageEngine::sync_storage_vault() {
     cloud::StorageVaultInfos vault_infos;
-    auto st = _meta_mgr->get_storage_vault_info(&vault_infos);
+    bool enable_storage_vault = false;
+    auto st = _meta_mgr->get_storage_vault_info(&vault_infos, &enable_storage_vault);
     if (!st.ok()) {
         LOG(WARNING) << "failed to get storage vault info. err=" << st;
         return;
@@ -363,7 +369,7 @@ void CloudStorageEngine::sync_storage_vault() {
     }
 
     if (auto& id = std::get<0>(vault_infos.back());
-        latest_fs() == nullptr || latest_fs()->id() != id) {
+        (latest_fs() == nullptr || latest_fs()->id() != id) && !enable_storage_vault) {
         set_latest_fs(get_filesystem(id));
     }
 }

be/src/common/config.cpp

@@ -95,6 +95,9 @@ DEFINE_String(mem_limit, "90%");
 // Soft memory limit as a fraction of hard memory limit.
 DEFINE_Double(soft_mem_limit_frac, "0.9");
 
+// Cache capacity reduce mem limit as a fraction of soft mem limit.
+DEFINE_mDouble(cache_capacity_reduce_mem_limit_frac, "0.6");
+
 // Schema change memory limit as a fraction of soft memory limit.
 DEFINE_Double(schema_change_mem_limit_frac, "0.6");
 
@@ -286,7 +289,7 @@ DEFINE_mInt32(exchg_buffer_queue_capacity_factor, "64");
 DEFINE_mInt64(memory_limitation_per_thread_for_schema_change_bytes, "2147483648");
 
 DEFINE_mInt32(cache_prune_interval_sec, "10");
-DEFINE_mInt32(cache_periodic_prune_stale_sweep_sec, "300");
+DEFINE_mInt32(cache_periodic_prune_stale_sweep_sec, "60");
 // the clean interval of tablet lookup cache
 DEFINE_mInt32(tablet_lookup_cache_stale_sweep_time_sec, "30");
 DEFINE_mInt32(point_query_row_cache_stale_sweep_time_sec, "300");
@@ -565,7 +568,7 @@ DEFINE_String(pprof_profile_dir, "${DORIS_HOME}/log");
 // for jeprofile in jemalloc
 DEFINE_mString(jeprofile_dir, "${DORIS_HOME}/log");
 DEFINE_mBool(enable_je_purge_dirty_pages, "true");
-DEFINE_mString(je_dirty_pages_mem_limit_percent, "5%");
+DEFINE_mString(je_dirty_pages_mem_limit_percent, "2%");
 
 // to forward compatibility, will be removed later
 DEFINE_mBool(enable_token_check, "true");
@@ -582,17 +585,12 @@ DEFINE_Int32(num_cores, "0");
 DEFINE_Bool(ignore_broken_disk, "false");
 
 // Sleep time in milliseconds between memory maintenance iterations
-DEFINE_mInt32(memory_maintenance_sleep_time_ms, "100");
+DEFINE_mInt32(memory_maintenance_sleep_time_ms, "20");
 
 // After full gc, no longer full gc and minor gc during sleep.
 // After minor gc, no minor gc during sleep, but full gc is possible.
 DEFINE_mInt32(memory_gc_sleep_time_ms, "500");
 
-// Sleep time in milliseconds between memtbale flush mgr refresh iterations
-DEFINE_mInt64(memtable_mem_tracker_refresh_interval_ms, "5");
-
-DEFINE_mInt64(wg_weighted_memory_ratio_refresh_interval_ms, "50");
-
 // percent of (active memtables size / all memtables size) when reach hard limit
 DEFINE_mInt32(memtable_hard_limit_active_percent, "50");
 

be/src/common/config.h

@@ -133,6 +133,9 @@ DECLARE_String(mem_limit);
 // Soft memory limit as a fraction of hard memory limit.
 DECLARE_Double(soft_mem_limit_frac);
 
+// Cache capacity reduce mem limit as a fraction of soft mem limit.
+DECLARE_mDouble(cache_capacity_reduce_mem_limit_frac);
+
 // Schema change memory limit as a fraction of soft memory limit.
 DECLARE_Double(schema_change_mem_limit_frac);
 
@@ -641,12 +644,6 @@ DECLARE_mInt32(memory_maintenance_sleep_time_ms);
 // After minor gc, no minor gc during sleep, but full gc is possible.
 DECLARE_mInt32(memory_gc_sleep_time_ms);
 
-// Sleep time in milliseconds between memtbale flush mgr memory refresh iterations
-DECLARE_mInt64(memtable_mem_tracker_refresh_interval_ms);
-
-// Sleep time in milliseconds between refresh iterations of workload group weighted memory ratio
-DECLARE_mInt64(wg_weighted_memory_ratio_refresh_interval_ms);
-
 // percent of (active memtables size / all memtables size) when reach hard limit
 DECLARE_mInt32(memtable_hard_limit_active_percent);