[Enhancement] metadata support LRU memory evict strategy in shared-nothing cluster (backport #48832)

be/src/common/config.h

@@ -538,6 +538,9 @@ CONF_Int64(compaction_memory_limit_per_worker, "2147483648"); // 2GB
 CONF_String(consistency_max_memory_limit, "10G");
 CONF_Int32(consistency_max_memory_limit_percent, "20");
 CONF_Int32(update_memory_limit_percent, "60");
+// Metadata cache limit for shared-nothing mode. Not working for PK table now.
+// Disable metadata cache when metadata_cache_memory_limit_percent <= 0.
+CONF_mInt32(metadata_cache_memory_limit_percent, "30"); // 30%
 
 // if `enable_retry_apply`, it apply failed due to some tolerable error(e.g. memory exceed limit)
 // the failed apply task will retry after `retry_apply_interval_second`

be/src/storage/CMakeLists.txt

@@ -120,6 +120,7 @@ add_library(Storage STATIC
     rowset/rowid_range_option.cpp
     rowset/rowset_meta.cpp
     rowset/horizontal_update_rowset_writer.cpp
+    rowset/metadata_cache.cpp
     task/engine_batch_load_task.cpp
     task/engine_checksum_task.cpp
     task/engine_clone_task.cpp

be/src/storage/rowset/metadata_cache.cpp

@@ -0,0 +1,52 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 "storage/rowset/metadata_cache.h"
+
+#include "storage/rowset/rowset.h"
+#include "util/lru_cache.h"
+
+namespace starrocks {
+
+MetadataCache::MetadataCache(size_t capacity) {
+    _cache.reset(new_lru_cache(capacity));
+}
+
+void MetadataCache::cache_rowset(Rowset* ptr) {
+    _insert(ptr->rowset_id_str(), ptr, ptr->segment_memory_usage());
+}
+
+void MetadataCache::evict_rowset(Rowset* ptr) {
+    _erase(ptr->rowset_id_str());
+}
+
+size_t MetadataCache::get_memory_usage() const {
+    return _cache->get_memory_usage();
+}
+
+void MetadataCache::_insert(const std::string& key, Rowset* ptr, size_t size) {
+    Cache::Handle* handle = _cache->insert(CacheKey(key), ptr, size, _cache_value_deleter);
+    _cache->release(handle);
+}
+
+void MetadataCache::_erase(const std::string& key) {
+    _cache->erase(CacheKey(key));
+}
+
+void MetadataCache::_cache_value_deleter(const CacheKey& /*key*/, void* value) {
+    // close this rowset, release metadata memory
+    reinterpret_cast<Rowset*>(value)->close();
+}
+
+} // namespace starrocks

be/src/storage/rowset/metadata_cache.h

@@ -0,0 +1,55 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 <memory>
+#include <mutex>
+#include <string_view>
+#include <variant>
+
+#include "gutil/macros.h"
+
+namespace starrocks {
+class Cache;
+class CacheKey;
+class Rowset;
+
+class MetadataCache {
+public:
+    explicit MetadataCache(size_t capacity);
+
+    ~MetadataCache() {}
+
+    DISALLOW_COPY_AND_MOVE(MetadataCache);
+
+    // will be called after rowset load metadata.
+    void cache_rowset(Rowset* ptr);
+
+    // evict this rowset manually, will be called before rowset destroy.
+    void evict_rowset(Rowset* ptr);
+
+    // Memory usage of lru cache
+    size_t get_memory_usage() const;
+
+private:
+    void _insert(const std::string& key, Rowset* ptr, size_t size);
+    void _erase(const std::string& key);
+    static void _cache_value_deleter(const CacheKey& /*key*/, void* value);
+
+    // LRU cache for metadata
+    std::unique_ptr<Cache> _cache;
+};
+
+} // namespace starrocks

be/src/storage/rowset/rowset.cpp

@@ -51,6 +51,7 @@
 #include "storage/inverted/index_descriptor.hpp"
 #include "storage/merge_iterator.h"
 #include "storage/projection_iterator.h"
+#include "storage/rowset/metadata_cache.h"
 #include "storage/rowset/rowid_range_option.h"
 #include "storage/rowset/short_key_range_option.h"
 #include "storage/storage_engine.h"
@@ -76,6 +77,13 @@ Rowset::Rowset(const TabletSchemaCSPtr& schema, std::string rowset_path, RowsetM
 }
 
 Rowset::~Rowset() {
+#ifndef BE_TEST
+    if (_keys_type != PRIMARY_KEYS) {
+        // ONLY support non-pk table now.
+        // evict rowset before destroy, in case this rowset no close yet.
+        StorageEngine::instance()->tablet_manager()->metadata_cache()->evict_rowset(this);
+    }
+#endif
     MEM_TRACKER_SAFE_RELEASE(GlobalEnv::GetInstance()->rowset_metadata_mem_tracker(), _mem_usage());
 }
 
@@ -176,6 +184,13 @@ Status Rowset::do_load() {
         }
         _segments.push_back(std::move(res).value());
     }
+#ifndef BE_TEST
+    if (config::metadata_cache_memory_limit_percent > 0 && _keys_type != PRIMARY_KEYS) {
+        // Add rowset to lru metadata cache for memory control.
+        // ONLY support non-pk table now.
+        StorageEngine::instance()->tablet_manager()->metadata_cache()->cache_rowset(this);
+    }
+#endif
     return Status::OK();
 }
 
@@ -609,6 +624,14 @@ void Rowset::do_close() {
     _segments.clear();
 }
 
+size_t Rowset::segment_memory_usage() {
+    size_t total = 0;
+    for (const auto& segment : _segments) {
+        total += segment->mem_usage();
+    }
+    return total;
+}
+
 class SegmentIteratorWrapper : public ChunkIterator {
 public:
     SegmentIteratorWrapper(std::shared_ptr<Rowset> rowset, ChunkIteratorPtr iter)

be/src/storage/rowset/rowset.h

@@ -375,6 +375,8 @@ class Rowset : public std::enable_shared_from_this<Rowset>, public BaseRowset {
 
     Status verify();
 
+    size_t segment_memory_usage();
+
 protected:
     friend class RowsetFactory;
 

be/src/storage/storage_engine.cpp

@@ -63,6 +63,7 @@
 #include "storage/memtable_flush_executor.h"
 #include "storage/publish_version_manager.h"
 #include "storage/replication_txn_manager.h"
+#include "storage/rowset/metadata_cache.h"
 #include "storage/rowset/rowset_meta.h"
 #include "storage/rowset/rowset_meta_manager.h"
 #include "storage/rowset/unique_rowset_id_generator.h"

be/src/storage/tablet_manager.cpp

@@ -50,6 +50,7 @@
 #include "storage/compaction_manager.h"
 #include "storage/data_dir.h"
 #include "storage/olap_common.h"
+#include "storage/rowset/metadata_cache.h"
 #include "storage/rowset/rowset_factory.h"
 #include "storage/rowset/rowset_writer.h"
 #include "storage/rowset/rowset_writer_context.h"
@@ -88,6 +89,13 @@ TabletManager::TabletManager(int64_t tablet_map_lock_shard_size)
           _last_update_stat_ms(0) {
     CHECK_GT(_tablets_shards.size(), 0) << "tablets shard count greater than 0";
     CHECK_EQ(_tablets_shards.size() & _tablets_shards_mask, 0) << "tablets shard count must be power of two";
+#ifndef BE_TEST
+    const int64_t process_limit = GlobalEnv::GetInstance()->process_mem_tracker()->limit();
+    const int32_t lru_cache_limit = process_limit * config::metadata_cache_memory_limit_percent / 100;
+    _metadata_cache = std::make_unique<MetadataCache>(lru_cache_limit);
+    REGISTER_GAUGE_STARROCKS_METRIC(metadata_cache_bytes_total,
+                                    [this]() { return _metadata_cache->get_memory_usage(); });
+#endif
 }
 
 Status TabletManager::_add_tablet_unlocked(const TabletSharedPtr& new_tablet, bool update_meta, bool force) {

be/src/storage/tablet_manager.h

@@ -61,6 +61,7 @@ namespace starrocks {
 class Tablet;
 class DataDir;
 struct TabletBasicInfo;
+class MetadataCache;
 
 // RowsetsAcqRel is a RAII wrapper for invocation of Rowset::acquire_readers and Rowset::release_readers
 class RowsetsAcqRel;
@@ -198,6 +199,8 @@ class TabletManager {
 
     Status generate_pk_dump();
 
+    MetadataCache* metadata_cache() const { return _metadata_cache.get(); }
+
 private:
     using TabletMap = std::unordered_map<int64_t, TabletSharedPtr>;
     using TabletSet = std::unordered_set<int64_t>;
@@ -297,6 +300,9 @@ class TabletManager {
     // context for compaction checker
     size_t _cur_shard = 0;
     std::unordered_set<int64_t> _shard_visited_tablet_ids;
+
+    // LRU cache for metadata
+    std::unique_ptr<MetadataCache> _metadata_cache;
 };
 
 inline bool TabletManager::LockTable::is_locked(int64_t tablet_id) {

be/src/util/starrocks_metrics.h

@@ -219,6 +219,9 @@ class StarRocksMetrics {
     // Accumulated time that task pends in the queue
     METRIC_DEFINE_INT_COUNTER(async_delta_writer_task_pending_duration_us, MetricUnit::MICROSECONDS);
 
+    // Metrics for metadata lru cache
+    METRIC_DEFINE_INT_GAUGE(metadata_cache_bytes_total, MetricUnit::BYTES);
+
     // Metrics for delta writer
     // Accumulated time that delta writer waits for memtable flush. It's part of
     // async_delta_writer_task_execute_duration_us

be/test/CMakeLists.txt

@@ -287,6 +287,7 @@ set(EXEC_FILES
         ./storage/rowset/cast_column_iterator_test.cpp
         ./storage/rowset/series_column_iterator_test.cpp
         ./storage/rowset/index_page_test.cpp
+        ./storage/rowset/metadata_cache_test.cpp
         ./storage/snapshot_meta_test.cpp
         ./storage/short_key_index_test.cpp
         ./storage/storage_types_test.cpp