Merge branch 'master' into recycle-bin-fix

Author: xy720

be/src/cloud/cloud_stream_load_executor.cpp

@@ -104,6 +104,11 @@ Status CloudStreamLoadExecutor::operate_txn_2pc(StreamLoadContext* ctx) {
 
 Status CloudStreamLoadExecutor::commit_txn(StreamLoadContext* ctx) {
     DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.block", DBUG_BLOCK);
+    DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.crash", {
+        LOG(INFO) << "debug point " << DP_NAME << " trigger crash";
+        volatile int* p = nullptr;
+        *p = 1;
+    });
     // forward to fe to excute commit transaction for MoW table
     if (ctx->is_mow_table() || !config::enable_stream_load_commit_txn_on_be ||
         ctx->load_type == TLoadType::ROUTINE_LOAD) {

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/debug_points.h"
 #include "util/stack_util.h"
 
 namespace doris {
@@ -148,7 +149,9 @@ CloudTabletMgr::CloudTabletMgr(CloudStorageEngine& engine)
           _tablet_map(std::make_unique<TabletMap>()),
           _cache(std::make_unique<LRUCachePolicy>(
                   CachePolicy::CacheType::CLOUD_TABLET_CACHE, config::tablet_cache_capacity,
-                  LRUCacheType::NUMBER, 0, config::tablet_cache_shards, false /*enable_prune*/)) {}
+                  LRUCacheType::NUMBER, /*sweep time*/ 0, config::tablet_cache_shards,
+                  /*element_count_capacity*/ 0, /*enable_prune*/ false,
+                  /*is_lru_k*/ false)) {}
 
 CloudTabletMgr::~CloudTabletMgr() = default;
 
@@ -163,6 +166,7 @@ Result<std::shared_ptr<CloudTablet>> CloudTabletMgr::get_tablet(int64_t tablet_i
                                                                 SyncRowsetStats* sync_stats,
                                                                 bool force_use_only_cached,
                                                                 bool cache_on_miss) {
+    DBUG_EXECUTE_IF("CloudTabletMgr::get_tablet.block", DBUG_BLOCK);
     // LRU value type. `Value`'s lifetime MUST NOT be longer than `CloudTabletMgr`
     class Value : public LRUCacheValueBase {
     public:

be/src/cloud/cloud_txn_delete_bitmap_cache.cpp

@@ -34,7 +34,9 @@ namespace doris {
 
 CloudTxnDeleteBitmapCache::CloudTxnDeleteBitmapCache(size_t size_in_bytes)
         : LRUCachePolicy(CachePolicy::CacheType::CLOUD_TXN_DELETE_BITMAP_CACHE, size_in_bytes,
-                         LRUCacheType::SIZE, 86400, 4),
+                         LRUCacheType::SIZE, /*stale_sweep_time_s*/ 86400, /*num_shards*/ 4,
+                         /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                         /*is_lru_k*/ false),
           _stop_latch(1) {}
 
 CloudTxnDeleteBitmapCache::~CloudTxnDeleteBitmapCache() {

be/src/cloud/injection_point_action.cpp

@@ -279,6 +279,17 @@ void set_return_error(const std::string& point, HttpRequest* req) {
     HttpChannel::send_reply(req, HttpStatus::OK, "OK");
 }
 
+void set_segfault(const std::string& point, HttpRequest* req) {
+    auto sp = SyncPoint::get_instance();
+    sp->set_call_back(point, [point](auto&&) {
+        LOG(INFO) << "injection point hit, point=" << point << " trigger segfault";
+        // Intentional null dereference to crash the BE for testing.
+        volatile int* p = nullptr;
+        *p = 1;
+    });
+    HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
 void handle_set(HttpRequest* req) {
     auto& point = req->param("name");
     if (point.empty()) {
@@ -302,6 +313,9 @@ void handle_set(HttpRequest* req) {
     } else if (behavior == "return_error") {
         set_return_error(point, req);
         return;
+    } else if (behavior == "segfault") {
+        set_segfault(point, req);
+        return;
     }
     HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "unknown behavior: " + behavior);
 }
@@ -377,13 +391,15 @@ InjectionPointAction::InjectionPointAction() = default;
 //                 which is an int, valid values can be found in status.h, e.g. -235 or -230,
 //                 if `code` is not present return Status::InternalError. Optional `probability`
 //                 determines the percentage of times to inject the error (default 100).
+// * segfault: dereference a null pointer to crash BE intentionally
 // ```
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=sleep&duration=${x_millsec}" # sleep x millisecs
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return" # return void
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_ok" # return ok
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error" # internal error
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error&code=${code}" # -235
 // curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=return_error&code=${code}&probability=50" # inject with 50% probability
+// curl "be_ip:http_port/api/injection_point/set?name=${injection_point_name}&behavior=segfault" # crash BE
 // ```
 void InjectionPointAction::handle(HttpRequest* req) {
     LOG(INFO) << "handle InjectionPointAction " << req->debug_string();

be/src/exec/schema_scanner/schema_file_cache_info_scanner.cpp

@@ -30,6 +30,7 @@ namespace doris {
 std::vector<SchemaScanner::ColumnDesc> SchemaFileCacheInfoScanner::_s_tbls_columns = {
         //   name,       type,          size,     is_null
         {"HASH", TYPE_STRING, sizeof(StringRef), true},
+        {"OFFSET", TYPE_BIGINT, sizeof(int64_t), true},
         {"TABLET_ID", TYPE_BIGINT, sizeof(int64_t), true},
         {"SIZE", TYPE_BIGINT, sizeof(int64_t), true},
         {"TYPE", TYPE_STRING, sizeof(StringRef), true},
@@ -68,7 +69,7 @@ Status SchemaFileCacheInfoScanner::_fill_block_impl(vectorized::Block* block) {
     }
 
     // Collect all cache entries from all file cache instances
-    std::vector<std::tuple<std::string, int64_t, int64_t, int, std::string>> cache_entries;
+    std::vector<std::tuple<std::string, int64_t, int64_t, int64_t, int, std::string>> cache_entries;
 
     // Get all cache instances using the public getter
     const auto& caches = file_cache_factory->get_caches();
@@ -116,7 +117,8 @@ Status SchemaFileCacheInfoScanner::_fill_block_impl(vectorized::Block* block) {
             std::string hash_str = key.hash.to_string();
 
             // Add to cache entries
-            cache_entries.emplace_back(hash_str, key.tablet_id, value.size, value.type, cache_path);
+            cache_entries.emplace_back(hash_str, static_cast<int64_t>(key.offset), key.tablet_id,
+                                       static_cast<int64_t>(value.size), value.type, cache_path);
 
             iterator->next();
         }
@@ -137,21 +139,21 @@ Status SchemaFileCacheInfoScanner::_fill_block_impl(vectorized::Block* block) {
 
         for (size_t row_idx = 0; row_idx < row_num; ++row_idx) {
             const auto& entry = cache_entries[row_idx];
-            const auto& [hash, tablet_id, size, type, cache_path] = entry;
+            const auto& [hash, offset, tablet_id, size, type, cache_path] = entry;
 
             if (col_desc.type == TYPE_STRING) {
                 switch (col_idx) {
                 case 0: // HASH
                     column_values[row_idx] = hash;
                     break;
-                case 3: // TYPE
+                case 4: // TYPE
                     column_values[row_idx] = doris::io::cache_type_to_string(
                             static_cast<doris::io::FileCacheType>(type));
                     break;
-                case 4:                          // REMOTE_PATH
+                case 5:                          // REMOTE_PATH
                     column_values[row_idx] = ""; // TODO: Implement remote path retrieval
                     break;
-                case 5: // CACHE_PATH
+                case 6: // CACHE_PATH
                     column_values[row_idx] = cache_path;
                     break;
                 default:
@@ -163,13 +165,16 @@ Status SchemaFileCacheInfoScanner::_fill_block_impl(vectorized::Block* block) {
                 datas[row_idx] = &str_refs[row_idx];
             } else if (col_desc.type == TYPE_BIGINT) {
                 switch (col_idx) {
-                case 1: // TABLET_ID
+                case 1: // OFFSET
+                    int64_vals[row_idx] = offset;
+                    break;
+                case 2: // TABLET_ID
                     int64_vals[row_idx] = tablet_id;
                     break;
-                case 2: // SIZE
+                case 3: // SIZE
                     int64_vals[row_idx] = size;
                     break;
-                case 6: // BE_ID
+                case 7: // BE_ID
                     int64_vals[row_idx] = ExecEnv::GetInstance()->cluster_info()->backend_id;
                     break;
                 default:

be/src/olap/page_cache.h

@@ -118,24 +118,27 @@ class StoragePageCache {
         DataPageCache(size_t capacity, uint32_t num_shards)
                 : LRUCachePolicy(CachePolicy::CacheType::DATA_PAGE_CACHE, capacity,
                                  LRUCacheType::SIZE, config::data_page_cache_stale_sweep_time_sec,
-                                 num_shards, DEFAULT_LRU_CACHE_ELEMENT_COUNT_CAPACITY, true, true) {
-        }
+                                 num_shards, /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                                 /*is lru-k*/ true) {}
     };
 
     class IndexPageCache : public LRUCachePolicy {
     public:
         IndexPageCache(size_t capacity, uint32_t num_shards)
                 : LRUCachePolicy(CachePolicy::CacheType::INDEXPAGE_CACHE, capacity,
                                  LRUCacheType::SIZE, config::index_page_cache_stale_sweep_time_sec,
-                                 num_shards) {}
+                                 num_shards, /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                                 /*is lru-k*/ false) {}
     };
 
     class PKIndexPageCache : public LRUCachePolicy {
     public:
         PKIndexPageCache(size_t capacity, uint32_t num_shards)
                 : LRUCachePolicy(CachePolicy::CacheType::PK_INDEX_PAGE_CACHE, capacity,
                                  LRUCacheType::SIZE,
-                                 config::pk_index_page_cache_stale_sweep_time_sec, num_shards) {}
+                                 config::pk_index_page_cache_stale_sweep_time_sec, num_shards,
+                                 /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                                 /*is lru-k*/ false) {}
     };
 
     static constexpr uint32_t kDefaultNumShards = 16;

be/src/olap/rowset/segment_v2/condition_cache.h

@@ -83,7 +83,9 @@ class ConditionCache : public LRUCachePolicy {
 
     ConditionCache(size_t capacity, uint32_t num_shards)
             : LRUCachePolicy(CachePolicy::CacheType::CONDITION_CACHE, capacity, LRUCacheType::SIZE,
-                             config::inverted_index_cache_stale_sweep_time_sec, num_shards) {}
+                             config::inverted_index_cache_stale_sweep_time_sec, num_shards,
+                             /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                             /*is_lru_k*/ true) {}
 
     bool lookup(const CacheKey& key, ConditionCacheHandle* handle);
 

be/src/olap/rowset/segment_v2/inverted_index_cache.h

@@ -106,16 +106,17 @@ class InvertedIndexSearcherCache {
                 : LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
                                  LRUCacheType::SIZE,
                                  config::inverted_index_cache_stale_sweep_time_sec, num_shards,
-                                 element_count_capacity, true) {}
+                                 element_count_capacity, /*enable_prune*/ true,
+                                 /*is lru k*/ false) {}
         InvertedIndexSearcherCachePolicy(size_t capacity, uint32_t num_shards,
                                          uint32_t element_count_capacity,
                                          CacheValueTimeExtractor cache_value_time_extractor,
                                          bool cache_value_check_timestamp)
-                : LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
-                                 LRUCacheType::SIZE,
-                                 config::inverted_index_cache_stale_sweep_time_sec, num_shards,
-                                 element_count_capacity, cache_value_time_extractor,
-                                 cache_value_check_timestamp, true) {}
+                : LRUCachePolicy(
+                          CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
+                          LRUCacheType::SIZE, config::inverted_index_cache_stale_sweep_time_sec,
+                          num_shards, element_count_capacity, cache_value_time_extractor,
+                          cache_value_check_timestamp, /*enable_prune*/ true, /*is lru k*/ false) {}
     };
     // Insert a cache entry by key.
     // And the cache entry will be returned in handle.
@@ -229,7 +230,9 @@ class InvertedIndexQueryCache : public LRUCachePolicy {
     InvertedIndexQueryCache(size_t capacity, uint32_t num_shards)
             : LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_QUERY_CACHE, capacity,
                              LRUCacheType::SIZE, config::inverted_index_cache_stale_sweep_time_sec,
-                             num_shards) {}
+                             num_shards,
+                             /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                             /*is_lru_k*/ true) {}
 
     bool lookup(const CacheKey& key, InvertedIndexQueryCacheHandle* handle);
 

be/src/olap/schema_cache.h

@@ -87,7 +87,9 @@ class SchemaCache : public LRUCachePolicy {
 
     SchemaCache(size_t capacity)
             : LRUCachePolicy(CachePolicy::CacheType::SCHEMA_CACHE, capacity, LRUCacheType::NUMBER,
-                             config::schema_cache_sweep_time_sec) {}
+                             config::schema_cache_sweep_time_sec, /*num shards*/ 32,
+                             /*element_count_capacity*/ 0, /*enable_prune*/ true,
+                             /*is lru-k*/ false) {}
 
 private:
     static constexpr char SCHEMA_DELIMITER = '-';

be/src/olap/segment_loader.h

@@ -83,10 +83,11 @@ class SegmentCache : public LRUCachePolicy {
     };
 
     SegmentCache(size_t memory_bytes_limit, size_t segment_num_limit)
-            : LRUCachePolicy(
-                      CachePolicy::CacheType::SEGMENT_CACHE, memory_bytes_limit, LRUCacheType::SIZE,
-                      config::tablet_rowset_stale_sweep_time_sec, DEFAULT_LRU_CACHE_NUM_SHARDS * 2,
-                      cast_set<uint32_t>(segment_num_limit), config::enable_segment_cache_prune) {}
+            : LRUCachePolicy(CachePolicy::CacheType::SEGMENT_CACHE, memory_bytes_limit,
+                             LRUCacheType::SIZE, config::tablet_rowset_stale_sweep_time_sec,
+                             /*num shards*/ 64,
+                             /*element count capacity */ cast_set<uint32_t>(segment_num_limit),
+                             config::enable_segment_cache_prune, /*is lru-k*/ true) {}
 
     // Lookup the given segment in the cache.
     // If the segment is found, the cache entry will be written into handle.