Merge branch 'master' into refactor_datacache_session_switchvar

Author: wenzhenghu

be/src/common/config.cpp

@@ -1201,6 +1201,11 @@ DEFINE_mDouble(file_cache_keep_base_compaction_output_min_hit_ratio, "0.7");
 // if difference below this threshold, we consider cache's progressive upgrading (2.0->3.0) successful
 DEFINE_mDouble(file_cache_meta_store_vs_file_system_diff_num_threshold, "0.3");
 DEFINE_mDouble(file_cache_keep_schema_change_output_min_hit_ratio, "0.7");
+DEFINE_mDouble(file_cache_leak_fs_to_meta_ratio_threshold, "1.3");
+DEFINE_mInt64(file_cache_leak_scan_interval_seconds, "86400");
+DEFINE_mInt32(file_cache_leak_scan_batch_files, "2048");
+DEFINE_mInt32(file_cache_leak_scan_pause_ms, "500");
+DEFINE_mInt64(file_cache_leak_grace_seconds, "3600");
 
 DEFINE_mInt64(file_cache_remove_block_qps_limit, "1000");
 DEFINE_mInt64(file_cache_background_gc_interval_ms, "100");

be/src/common/config.h

@@ -1230,6 +1230,11 @@ DECLARE_mBool(enable_file_cache_adaptive_write);
 DECLARE_mDouble(file_cache_keep_base_compaction_output_min_hit_ratio);
 DECLARE_mDouble(file_cache_meta_store_vs_file_system_diff_num_threshold);
 DECLARE_mDouble(file_cache_keep_schema_change_output_min_hit_ratio);
+DECLARE_mDouble(file_cache_leak_fs_to_meta_ratio_threshold);
+DECLARE_mInt64(file_cache_leak_scan_interval_seconds);
+DECLARE_mInt32(file_cache_leak_scan_batch_files);
+DECLARE_mInt32(file_cache_leak_scan_pause_ms);
+DECLARE_mInt64(file_cache_leak_grace_seconds);
 DECLARE_mInt64(file_cache_remove_block_qps_limit);
 DECLARE_mInt64(file_cache_background_gc_interval_ms);
 DECLARE_mInt64(file_cache_background_block_lru_update_interval_ms);

be/src/io/cache/cache_block_meta_store.cpp

@@ -332,6 +332,33 @@ std::unique_ptr<BlockMetaIterator> CacheBlockMetaStore::get_all() {
     return std::unique_ptr<BlockMetaIterator>(new RocksDBIterator(iter));
 }
 
+size_t CacheBlockMetaStore::approximate_entry_count() const {
+    if (!_db) {
+        LOG(WARNING) << "Database not initialized when counting entries";
+        return 0;
+    }
+
+    rocksdb::ReadOptions read_options;
+    std::unique_ptr<rocksdb::Iterator> iter(
+            _db->NewIterator(read_options, _file_cache_meta_cf_handle.get()));
+    if (!iter) {
+        LOG(WARNING) << "Failed to create iterator when counting entries";
+        return 0;
+    }
+
+    size_t count = 0;
+    for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+        ++count;
+    }
+
+    if (!iter->status().ok()) {
+        LOG(WARNING) << "Iterator encountered error when counting entries: "
+                     << iter->status().ToString();
+    }
+
+    return count;
+}
+
 void CacheBlockMetaStore::delete_key(const BlockMetaKey& key) {
     std::string key_str = serialize_key(key);
 

be/src/io/cache/cache_block_meta_store.h

@@ -111,6 +111,9 @@ class CacheBlockMetaStore {
     // Get the approximate size of the write queue
     size_t get_write_queue_size() const;
 
+    // Count entries stored in rocksdb (ignoring pending writes)
+    size_t approximate_entry_count() const;
+
 private:
     void async_write_worker();