Merge branch 'master' into doc_for_recover

README.md

@@ -59,12 +59,9 @@ Apache Doris is an easy-to-use, high-performance and real-time analytical databa
 
 All this makes Apache Doris an ideal tool for scenarios including report analysis, ad-hoc query, unified data warehouse, and data lake query acceleration. On Apache Doris, users can build various applications, such as user behavior analysis, AB test platform, log retrieval analysis, user portrait analysis, and order analysis.
 
-🎉 Version 2.1.4 released now. Check out the 🔗[Release Notes](https://doris.apache.org/docs/releasenotes/release-2.1.4) here. The 2.1 verison delivers exceptional performance with 100% higher out-of-the-box queries proven by TPC-DS 1TB tests, enhanced data lake analytics that are 4-6 times speedier than Trino and Spark, solid support for semi-structured data analysis with new Variant types and suite of analytical functions, asynchronous materialized views for query acceleration, optimized real-time writing at scale, and better workload management with stability and runtime SQL resource tracking.
+🎉 Check out the 🔗[All releases](https://doris.apache.org/docs/releasenotes/all-release), where you'll find a chronological summary of Apache Doris versions released over the past year.
 
-
-🎉 Version 2.0.12 is now released ! This fully evolved and stable release is ready for all users to upgrade. Check out the 🔗[Release Notes](https://doris.apache.org/docs/2.0/releasenotes/release-2.0.12) here. 
-
-👀 Have a look at the 🔗[Official Website](https://doris.apache.org/) for a comprehensive list of Apache Doris's core features, blogs and user cases.
+👀 Explore the 🔗[Official Website](https://doris.apache.org/) to discover Apache Doris's core features, blogs, and user cases in detail.
 
 ## 📈 Usage Scenarios
 

be/src/agent/workload_group_listener.cpp

@@ -17,6 +17,7 @@
 
 #include "agent/workload_group_listener.h"
 
+#include "runtime/exec_env.h"
 #include "runtime/workload_group/workload_group.h"
 #include "runtime/workload_group/workload_group_manager.h"
 #include "util/mem_info.h"

be/src/agent/workload_group_listener.h

@@ -20,10 +20,11 @@
 #include <glog/logging.h>
 
 #include "agent/topic_listener.h"
-#include "runtime/exec_env.h"
 
 namespace doris {
 
+class ExecEnv;
+
 class WorkloadGroupListener : public TopicListener {
 public:
     ~WorkloadGroupListener() {}

be/src/cloud/cloud_tablet.cpp

@@ -54,6 +54,7 @@ namespace doris {
 using namespace ErrorCode;
 
 static constexpr int COMPACTION_DELETE_BITMAP_LOCK_ID = -1;
+static constexpr int LOAD_INITIATOR_ID = -1;
 
 CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)
         : BaseTablet(std::move(tablet_meta)), _engine(engine) {}
@@ -504,13 +505,19 @@ Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer(
 Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_writer(
         const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
         int64_t txn_expiration) {
-    if (rowset.rowset_meta()->rowset_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE) [[unlikely]] {
-        // May cause the segment files generated by the transient rowset writer unable to be
-        // recycled, see `CloudRowsetWriter::build` for detail.
-        LOG(WARNING) << "Wrong rowset state: " << rowset.rowset_meta()->rowset_state();
-        DCHECK(false) << rowset.rowset_meta()->rowset_state();
+    if (rowset.rowset_meta_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE &&
+        rowset.rowset_meta_state() != RowsetStatePB::COMMITTED) [[unlikely]] {
+        auto msg = fmt::format(
+                "wrong rowset state when create_transient_rowset_writer, rowset state should be "
+                "BEGIN_PARTIAL_UPDATE or COMMITTED, but found {}, rowset_id={}, tablet_id={}",
+                RowsetStatePB_Name(rowset.rowset_meta_state()), rowset.rowset_id().to_string(),
+                tablet_id());
+        // see `CloudRowsetWriter::build` for detail.
+        // if this is in a retry task, the rowset state may have been changed to RowsetStatePB::COMMITTED
+        // in `RowsetMeta::merge_rowset_meta()` in previous trials.
+        LOG(WARNING) << msg;
+        DCHECK(false) << msg;
     }
-
     RowsetWriterContext context;
     context.rowset_state = PREPARED;
     context.segments_overlap = OVERLAPPING;
@@ -719,8 +726,8 @@ Status CloudTablet::save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t tx
     }
 
     auto ms_lock_id = lock_id == -1 ? txn_id : lock_id;
-    RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(
-            *this, ms_lock_id, COMPACTION_DELETE_BITMAP_LOCK_ID, new_delete_bitmap.get()));
+    RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(*this, ms_lock_id, LOAD_INITIATOR_ID,
+                                                            new_delete_bitmap.get()));
 
     // store the delete bitmap with sentinel marks in txn_delete_bitmap_cache because if the txn is retried for some reason,
     // it will use the delete bitmap from txn_delete_bitmap_cache when re-calculating the delete bitmap, during which it will do

be/src/cloud/cloud_tablet_hotspot.cpp

@@ -57,18 +57,55 @@ TabletHotspot::~TabletHotspot() {
     }
 }
 
-struct MapKeyHash {
-    int64_t operator()(const std::pair<int64_t, int64_t>& key) const {
-        return std::hash<int64_t> {}(key.first) + std::hash<int64_t> {}(key.second);
+void get_return_partitions(
+        const std::unordered_map<TabletHotspotMapKey,
+                                 std::unordered_map<int64_t, TabletHotspotMapValue>, MapKeyHash>&
+                hot_partition,
+        const std::unordered_map<TabletHotspotMapKey,
+                                 std::unordered_map<int64_t, TabletHotspotMapValue>, MapKeyHash>&
+                last_hot_partition,
+        std::vector<THotTableMessage>* hot_tables, int& return_partitions, int N) {
+    for (const auto& [key, partition_to_value] : hot_partition) {
+        THotTableMessage msg;
+        msg.table_id = key.first;
+        msg.index_id = key.second;
+        for (const auto& [partition_id, value] : partition_to_value) {
+            if (return_partitions > N) {
+                return;
+            }
+            auto last_value_iter = last_hot_partition.find(key);
+            if (last_value_iter != last_hot_partition.end()) {
+                auto last_partition_iter = last_value_iter->second.find(partition_id);
+                if (last_partition_iter != last_value_iter->second.end()) {
+                    const auto& last_value = last_partition_iter->second;
+                    if (std::abs(static_cast<int64_t>(value.qpd) -
+                                 static_cast<int64_t>(last_value.qpd)) < 5 &&
+                        std::abs(static_cast<int64_t>(value.qpw) -
+                                 static_cast<int64_t>(last_value.qpw)) < 10 &&
+                        std::abs(static_cast<int64_t>(value.last_access_time) -
+                                 static_cast<int64_t>(last_value.last_access_time)) < 60) {
+                        LOG(INFO) << "skip partition_id=" << partition_id << " qpd=" << value.qpd
+                                  << " qpw=" << value.qpw
+                                  << " last_access_time=" << value.last_access_time
+                                  << " last_qpd=" << last_value.qpd
+                                  << " last_qpw=" << last_value.qpw
+                                  << " last_access_time=" << last_value.last_access_time;
+                        continue;
+                    }
+                }
+            }
+            THotPartition hot_partition;
+            hot_partition.__set_partition_id(partition_id);
+            hot_partition.__set_query_per_day(value.qpd);
+            hot_partition.__set_query_per_week(value.qpw);
+            hot_partition.__set_last_access_time(value.last_access_time);
+            msg.hot_partitions.push_back(hot_partition);
+            return_partitions++;
+        }
+        msg.__isset.hot_partitions = !msg.hot_partitions.empty();
+        hot_tables->push_back(std::move(msg));
     }
-};
-struct TabletHotspotMapValue {
-    uint64_t qpd = 0; // query per day
-    uint64_t qpw = 0; // query per week
-    int64_t last_access_time;
-};
-
-using TabletHotspotMapKey = std::pair<int64_t, int64_t>;
+}
 
 void TabletHotspot::get_top_n_hot_partition(std::vector<THotTableMessage>* hot_tables) {
     // map<pair<table_id, index_id>, map<partition_id, value>> for day
@@ -108,33 +145,14 @@ void TabletHotspot::get_top_n_hot_partition(std::vector<THotTableMessage>* hot_t
     });
     constexpr int N = 50;
     int return_partitions = 0;
-    auto get_return_partitions =
-            [=, &return_partitions](
-                    const std::unordered_map<TabletHotspotMapKey,
-                                             std::unordered_map<int64_t, TabletHotspotMapValue>,
-                                             MapKeyHash>& hot_partition) {
-                for (const auto& [key, partition_to_value] : hot_partition) {
-                    THotTableMessage msg;
-                    msg.table_id = key.first;
-                    msg.index_id = key.second;
-                    for (const auto& [partition_id, value] : partition_to_value) {
-                        if (return_partitions > N) {
-                            return;
-                        }
-                        THotPartition hot_partition;
-                        hot_partition.__set_partition_id(partition_id);
-                        hot_partition.__set_query_per_day(value.qpd);
-                        hot_partition.__set_query_per_week(value.qpw);
-                        hot_partition.__set_last_access_time(value.last_access_time);
-                        msg.hot_partitions.push_back(hot_partition);
-                        return_partitions++;
-                    }
-                    msg.__isset.hot_partitions = !msg.hot_partitions.empty();
-                    hot_tables->push_back(std::move(msg));
-                }
-            };
-    get_return_partitions(day_hot_partitions);
-    get_return_partitions(week_hot_partitions);
+
+    get_return_partitions(day_hot_partitions, _last_day_hot_partitions, hot_tables,
+                          return_partitions, N);
+    get_return_partitions(week_hot_partitions, _last_week_hot_partitions, hot_tables,
+                          return_partitions, N);
+
+    _last_day_hot_partitions = std::move(day_hot_partitions);
+    _last_week_hot_partitions = std::move(week_hot_partitions);
 }
 
 void HotspotCounter::make_dot_point() {

be/src/cloud/cloud_tablet_hotspot.h

@@ -49,6 +49,19 @@ struct HotspotCounter {
 };
 
 using HotspotCounterPtr = std::shared_ptr<HotspotCounter>;
+using TabletHotspotMapKey = std::pair<int64_t, int64_t>;
+
+struct TabletHotspotMapValue {
+    uint64_t qpd = 0; // query per day
+    uint64_t qpw = 0; // query per week
+    int64_t last_access_time;
+};
+
+struct MapKeyHash {
+    int64_t operator()(const std::pair<int64_t, int64_t>& key) const {
+        return std::hash<int64_t> {}(key.first) + std::hash<int64_t> {}(key.second);
+    }
+};
 
 class TabletHotspot {
 public:
@@ -71,6 +84,12 @@ class TabletHotspot {
     bool _closed {false};
     std::mutex _mtx;
     std::condition_variable _cond;
+    std::unordered_map<TabletHotspotMapKey, std::unordered_map<int64_t, TabletHotspotMapValue>,
+                       MapKeyHash>
+            _last_day_hot_partitions;
+    std::unordered_map<TabletHotspotMapKey, std::unordered_map<int64_t, TabletHotspotMapValue>,
+                       MapKeyHash>
+            _last_week_hot_partitions;
 };
 
 } // namespace doris

be/src/common/config.cpp

@@ -1404,6 +1404,8 @@ DEFINE_Bool(enable_table_size_correctness_check, "false");
 DEFINE_Bool(force_regenerate_rowsetid_on_start_error, "false");
 DEFINE_mBool(enable_sleep_between_delete_cumu_compaction, "false");
 
+DEFINE_mInt32(compaction_num_per_round, "1");
+
 // clang-format off
 #ifdef BE_TEST
 // test s3

be/src/common/config.h

@@ -1490,6 +1490,8 @@ DECLARE_Bool(enable_table_size_correctness_check);
 // Enable sleep 5s between delete cumulative compaction.
 DECLARE_mBool(enable_sleep_between_delete_cumu_compaction);
 
+DECLARE_mInt32(compaction_num_per_round);
+
 #ifdef BE_TEST
 // test s3
 DECLARE_String(test_s3_resource);

be/src/common/daemon.cpp

@@ -230,6 +230,11 @@ void refresh_memory_state_after_memory_change() {
 }
 
 void refresh_cache_capacity() {
+    if (doris::GlobalMemoryArbitrator::cache_adjust_capacity_notify.load(
+                std::memory_order_relaxed)) {
+        // the last cache capacity adjustment has not been completed.
+        return;
+    }
     if (refresh_cache_capacity_sleep_time_ms <= 0) {
         auto cache_capacity_reduce_mem_limit = int64_t(
                 doris::MemInfo::soft_mem_limit() * config::cache_capacity_reduce_mem_limit_frac);
@@ -247,6 +252,8 @@ void refresh_cache_capacity() {
                     new_cache_capacity_adjust_weighted;
             doris::GlobalMemoryArbitrator::notify_cache_adjust_capacity();
             refresh_cache_capacity_sleep_time_ms = config::memory_gc_sleep_time_ms;
+        } else {
+            refresh_cache_capacity_sleep_time_ms = 0;
         }
     }
     refresh_cache_capacity_sleep_time_ms -= config::memory_maintenance_sleep_time_ms;

be/src/common/status.h

@@ -293,7 +293,8 @@ namespace ErrorCode {
     E(ENTRY_NOT_FOUND, -7002, false);                        \
     E(INVALID_TABLET_STATE, -7211, false);                   \
     E(ROWSETS_EXPIRED, -7311, false);                        \
-    E(CGROUP_ERROR, -7411, false);
+    E(CGROUP_ERROR, -7411, false);                           \
+    E(FATAL_ERROR, -7412, false);
 
 // Define constexpr int error_code_name = error_code_value
 #define M(NAME, ERRORCODE, ENABLESTACKTRACE) constexpr int NAME = ERRORCODE;
@@ -446,6 +447,14 @@ class [[nodiscard]] Status {
 
     static Status OK() { return {}; }
 
+    template <bool stacktrace = true, typename... Args>
+    static Status FatalError(std::string_view msg, Args&&... args) {
+#ifndef NDEBUG
+        LOG(FATAL) << fmt::format(msg, std::forward<Args>(args)...);
+#endif
+        return Error<ErrorCode::FATAL_ERROR, stacktrace>(msg, std::forward<Args>(args)...);
+    }
+
 // default have stacktrace. could disable manually.
 #define ERROR_CTOR(name, code)                                                       \
     template <bool stacktrace = true, typename... Args>                              \
@@ -570,7 +579,7 @@ class [[nodiscard]] Status {
 // and another thread is call to_string method, it may core, because the _err_msg is an unique ptr and
 // it is deconstructed during copy method.
 // And also we could not use lock, because we need get status frequently to check if it is cancelled.
-// The defaule value is ok.
+// The default value is ok.
 class AtomicStatus {
 public:
     AtomicStatus() : error_st_(Status::OK()) {}

be/src/common/version_internal.cpp

@@ -34,6 +34,9 @@ int doris_build_version_minor() {
 int doris_build_version_patch() {
     return DORIS_BUILD_VERSION_PATCH;
 }
+int doris_build_version_hotfix() {
+    return DORIS_BUILD_VERSION_HOTFIX;
+}
 const char* doris_build_version_rc_version() {
     return DORIS_BUILD_VERSION_RC_VERSION;
 }
@@ -56,4 +59,4 @@ const char* doris_build_info() {
 
 } // namespace version
 
-} // namespace doris
+} // namespace doris

be/src/common/version_internal.h

@@ -24,6 +24,7 @@ extern const char* doris_build_version_prefix();
 extern int doris_build_version_major();
 extern int doris_build_version_minor();
 extern int doris_build_version_patch();
+extern int doris_build_version_hotfix();
 extern const char* doris_build_version_rc_version();
 
 extern const char* doris_build_version();
@@ -34,4 +35,4 @@ extern const char* doris_build_info();
 
 } // namespace version
 
-} // namespace doris
+} // namespace doris

be/src/exprs/runtime_filter.cpp

@@ -498,12 +498,12 @@ class RuntimePredicateWrapper {
         switch (_filter_type) {
         case RuntimeFilterType::IN_FILTER: {
             if (!_context->hybrid_set) {
-                _context->ignored = true;
+                set_ignored();
                 return Status::OK();
             }
             _context->hybrid_set->insert(wrapper->_context->hybrid_set.get());
             if (_max_in_num >= 0 && _context->hybrid_set->size() >= _max_in_num) {
-                _context->ignored = true;
+                set_ignored();
                 // release in filter
                 _context->hybrid_set.reset();
             }
@@ -1337,7 +1337,7 @@ void IRuntimeFilter::set_synced_size(uint64_t global_size) {
 }
 
 void IRuntimeFilter::set_ignored() {
-    _wrapper->_context->ignored = true;
+    _wrapper->set_ignored();
 }
 
 bool IRuntimeFilter::get_ignored() {

be/src/gutil/strings/escaping.cc

@@ -10,6 +10,8 @@
 #include <limits>
 #include <ostream>
 
+#include "common/exception.h"
+
 using std::numeric_limits;
 #include <vector>
 
@@ -1084,7 +1086,8 @@ int Base64UnescapeInternal(const char* src, int szsrc, char* dest, int szdest,
 
     default:
         // state should have no other values at this point.
-        LOG(FATAL) << "This can't happen; base64 decoder state = " << state;
+        throw doris::Exception(
+                doris::Status::FatalError("This can't happen; base64 decoder state = {}", state));
     }
 
     // The remainder of the string should be all whitespace, mixed with

be/src/gutil/strings/numbers.cc

@@ -19,6 +19,8 @@
 #include <limits>
 #include <ostream>
 
+#include "common/exception.h"
+
 using std::numeric_limits;
 #include <string>
 
@@ -772,8 +774,8 @@ uint64 atoi_kmgt(const char* s) {
             scale = GG_ULONGLONG(1) << 40;
             break;
         default:
-            LOG(FATAL) << "Invalid mnemonic: `" << c << "';"
-                       << " should be one of `K', `M', `G', and `T'.";
+            throw doris::Exception(doris::Status::FatalError(
+                    "Invalid mnemonic: `{}'; should be one of `K', `M', `G', and `T'.", c));
         }
     }
     return n * scale;