use bloom filter, add event listener to collect stats of compaction, improve update rocksdb options

src/common/base/Configuration.cpp

@@ -8,6 +8,10 @@
 
 namespace nebula {
 
+Configuration::Configuration() {
+    content_ = std::make_unique<folly::dynamic>(folly::dynamic::object());
+}
+
 Configuration::Configuration(folly::dynamic content) {
     CHECK(content.isObject()) << "The content is not a valid configuration";
     content_ = std::make_unique<folly::dynamic>(std::move(content));
@@ -160,7 +164,7 @@ Status Configuration::fetchAsSubConf(const char *key, Configuration &subconf) co
 }
 
 
-Status Configuration::updateStringField(const char* key, const std::string& val) {
+Status Configuration::upsertStringField(const char* key, const std::string& val) {
     DCHECK(content_ != nullptr);
     auto iter = content_->find(key);
     if (iter == content_->items().end() || iter->second.isString()) {

src/common/base/Configuration.h

@@ -19,7 +19,7 @@ namespace nebula {
 
 class Configuration final {
 public:
-    Configuration() = default;
+    Configuration();
     explicit Configuration(folly::dynamic content);
 
     ~Configuration() = default;
@@ -58,7 +58,7 @@ class Configuration final {
 
     Status MUST_USE_RESULT fetchAsSubConf(const char *key, Configuration &val) const;
 
-    Status MUST_USE_RESULT updateStringField(const char* key, const std::string& val);
+    Status MUST_USE_RESULT upsertStringField(const char* key, const std::string& val);
 
     // Iterate through every key in the configuration
     Status forEachKey(std::function<void(const std::string&)> processor) const;

src/kvstore/EventListner.h

@@ -0,0 +1,28 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "base/Base.h"
+#include "rocksdb/db.h"
+#include "rocksdb/listener.h"
+
+namespace nebula {
+namespace kvstore {
+
+class EventListener : public rocksdb::EventListener {
+public:
+    void OnCompactionCompleted(rocksdb::DB*, const rocksdb::CompactionJobInfo& info) override {
+        LOG(INFO) << "Rocksdb compact column family: " << info.cf_name
+                  << " because of " << static_cast<int32_t>(info.compaction_reason)
+                  << ", status: " << info.status.ToString()
+                  << ", compacted " << info.input_files.size()
+                  << " files into " << info.output_files.size()
+                  << ", base level is " << info.base_input_level
+                  << ", output level is " << info.output_level;
+    }
+};
+
+}  // namespace kvstore
+}  // namespace nebula

src/kvstore/NebulaStore.cpp

@@ -627,6 +627,7 @@ ResultCode NebulaStore::compact(GraphSpaceID spaceId) {
 
     auto code = ResultCode::SUCCEEDED;
     std::vector<std::thread> threads;
+    LOG(INFO) << "Space " << spaceId << " start compaction.";
     for (auto& engine : space->engines_) {
         threads.emplace_back(std::thread([&engine, &code] {
             auto ret = engine->compact();
@@ -640,6 +641,7 @@ ResultCode NebulaStore::compact(GraphSpaceID spaceId) {
     for (auto& t : threads) {
         t.join();
     }
+    LOG(INFO) << "Space " << spaceId << " compaction done.";
     return code;
 }
 

src/kvstore/PartManager.cpp

@@ -96,25 +96,19 @@ void MetaServerBasedPartManager::onSpaceOptionUpdated(
         "disable_auto_compactions",
         "max_write_buffer_number",
         // TODO: write_buffer_size will cause rocksdb crash
-        // "write_buffer_size",
-        "compression",
+        "write_buffer_size",
         "level0_file_num_compaction_trigger",
         "level0_slowdown_writes_trigger",
         "level0_stop_writes_trigger",
         "target_file_size_base",
         "target_file_size_multiplier",
         "max_bytes_for_level_base",
         "max_bytes_for_level_multiplier",
-        "ttl",
-        "block_size",
-        "block_restart_interval"
     };
     static std::unordered_set<std::string> supportedDbOpt = {
         "max_total_wal_size",
         "delete_obsolete_files_period_micros",
         "max_background_jobs",
-        "base_background_compactions",
-        "max_background_compactions",
         "stats_dump_period_sec",
         "compaction_readahead_size",
         "writable_file_max_buffer_size",

src/kvstore/RocksEngineConfig.cpp

@@ -6,11 +6,13 @@
 
 #include "base/Base.h"
 #include "kvstore/RocksEngineConfig.h"
+#include "kvstore/EventListner.h"
 #include "rocksdb/db.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/utilities/options_util.h"
 #include "rocksdb/slice_transform.h"
+#include "rocksdb/filter_policy.h"
 #include "base/Configuration.h"
 
 // [WAL]
@@ -44,6 +46,7 @@ DEFINE_int32(rocksdb_batch_size,
 DEFINE_int64(rocksdb_block_cache, 1024,
              "The default block cache size used in BlockBasedTable. The unit is MB");
 
+DEFINE_bool(enable_partitioned_index_filter, false, "True for partitioned index filters");
 
 namespace nebula {
 namespace kvstore {
@@ -62,6 +65,7 @@ rocksdb::Status initRocksdbOptions(rocksdb::Options &baseOpts) {
     if (!s.ok()) {
         return s;
     }
+    dbOpts.listeners.emplace_back(new EventListener());
 
     std::unordered_map<std::string, std::string> cfOptsMap;
     if (!loadOptionsMap(cfOptsMap, FLAGS_rocksdb_column_family_options)) {
@@ -87,6 +91,15 @@ rocksdb::Status initRocksdbOptions(rocksdb::Options &baseOpts) {
     static std::shared_ptr<rocksdb::Cache> blockCache
         = rocksdb::NewLRUCache(FLAGS_rocksdb_block_cache * 1024 * 1024);
     bbtOpts.block_cache = blockCache;
+    bbtOpts.filter_policy.reset(rocksdb::NewBloomFilterPolicy(10, false));
+    if (FLAGS_enable_partitioned_index_filter) {
+        bbtOpts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
+        bbtOpts.partition_filters = true;
+        bbtOpts.cache_index_and_filter_blocks = true;
+        bbtOpts.cache_index_and_filter_blocks_with_high_priority = true;
+        bbtOpts.pin_l0_filter_and_index_blocks_in_cache =
+            baseOpts.compaction_style == rocksdb::CompactionStyle::kCompactionStyleLevel;
+    }
     baseOpts.table_factory.reset(NewBlockBasedTableFactory(bbtOpts));
     baseOpts.create_if_missing = true;
     return s;

src/kvstore/raftex/RaftPart.cpp

@@ -28,7 +28,7 @@ DEFINE_uint64(raft_snapshot_timeout, 60 * 5, "Max seconds between two snapshot r
 
 DEFINE_uint32(max_batch_size, 256, "The max number of logs in a batch");
 
-DEFINE_int32(wal_ttl, 86400, "Default wal ttl");
+DEFINE_int32(wal_ttl, 14400, "Default wal ttl");
 DEFINE_int64(wal_file_size, 16 * 1024 * 1024, "Default wal file size");
 DEFINE_int32(wal_buffer_size, 8 * 1024 * 1024, "Default wal buffer size");
 DEFINE_int32(wal_buffer_num, 2, "Default wal buffer number");

src/meta/processors/configMan/SetConfigProcessor.cpp

@@ -14,7 +14,7 @@ std::unordered_set<std::string> SetConfigProcessor::mutableFields_ = {
     // rocksdb_column_family_options
     "disable_auto_compactions",
     // TODO: write_buffer_size will cause rocksdb crash
-    // "write_buffer_size",
+    "write_buffer_size",
     "max_write_buffer_number",
     "level0_file_num_compaction_trigger",
     "level0_slowdown_writes_trigger",
@@ -23,14 +23,11 @@ std::unordered_set<std::string> SetConfigProcessor::mutableFields_ = {
     "target_file_size_multiplier",
     "max_bytes_for_level_base",
     "max_bytes_for_level_multiplier",
-    "ttl",
 
     // rocksdb_db_options
     "max_total_wal_size",
     "delete_obsolete_files_period_micros",
     "max_background_jobs",
-    "base_background_compactions",
-    "max_background_compactions",
     "stats_dump_period_sec",
     "compaction_readahead_size",
     "writable_file_max_buffer_size",
@@ -141,9 +138,6 @@ cpp2::ErrorCode SetConfigProcessor::setNestedConfig(const cpp2::ConfigModule& mo
     }
 
     Configuration conf;
-    auto confRet = conf.parseFromString(item.get_value());
-    CHECK(confRet.ok());
-
     std::vector<std::string> updateFields;
     folly::split(",", updateList, updateFields, true);
     bool updated = false;
@@ -156,11 +150,12 @@ cpp2::ErrorCode SetConfigProcessor::setNestedConfig(const cpp2::ConfigModule& mo
         auto key = field.substr(0, pos);
         auto value = field.substr(pos + 1);
         // TODO: Maybe need to handle illegal value here
-        if (mutableFields_.count(key) && conf.updateStringField(key.c_str(), value).ok()) {
+        if (!conf.upsertStringField(key.c_str(), value).ok()) {
+            LOG(ERROR) << "Update configs failed: " << key;
+            return cpp2::ErrorCode::E_UNSUPPORTED;
+        }
+        if (mutableFields_.count(key)) {
             updated = true;
-        } else {
-            LOG(ERROR) << "Unsupported configs " << key;
-            return cpp2::ErrorCode::E_NOT_FOUND;
         }
     }