Feature: The WAL command can be dynamically disabled

docs/ops/APIDifference.md

@@ -126,6 +126,26 @@ slaveof命令允许通过指定write2file(binlog)的文件名称及同步位置
 
 ### diskrecovery 
 Pika 原创命令，功能为当磁盘意外写满后，RocksDB 会进入写保护状态，当我们将空间调整为充足空间时，这个命令可以将 RocksDB 的写保护状态解除，变为可以继续写的状态, 避免了 Pika 因为磁盘写满后需要重启才能恢复写的情况，执行成功时返回 OK，如果当前磁盘空间依然不足，执行这个命令返回`"The available disk capacity is insufficient`，该命令执行时不需要额外参数，只需要执行 diskrecovery 即可。
-
+```bash
+> diskrecovery 
+> OK
+```
 ### clearreplicationid
-Pika 原创命令，功能是清空 Pika 实例的 replicationid 的值并持久化到配置文件中
+Pika 原创命令，功能是清空 Pika 实例的 replicationid 的值并持久化到配置文件中
+```bash
+> clearreplicationid
+> OK
+```
+
+### DisableWal
+我们可以根据 disablewal (true/false) 接受两个参数来决定 WAL 选项的开关，true 表示写 WAL 功能关闭，false 表示写 WAL 功能开启，需要注意的是写 WAL 默认是开启的 
+```bash
+> disablewal true  // 写 WAL 关闭
+> OK
+> disablewal false  // 写 WAL 开启
+> OK
+> disablewal asdfs // 识别到异常参数
+> (error) ERR Invalid parameter
+> disable false dasfasd // 参数数目异常
+> (error) ERR wrong number of arguments for 'disablewal' command
+```

include/pika_admin.h

@@ -475,6 +475,18 @@ class ClearReplicationIDCmd : public Cmd {
   void DoInitial() override;
 };
 
+class DisableWalCmd : public Cmd {
+ public:
+  DisableWalCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+  void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
+  void Merge() override{};
+  Cmd* Clone() override { return new DisableWalCmd(*this); }
+
+ private:
+  void DoInitial() override;
+};
+
 #ifdef WITH_COMMAND_DOCS
 class CommandCmd : public Cmd {
  public:

include/pika_command.h

@@ -52,6 +52,7 @@ const std::string kCmdNameHello = "hello";
 const std::string kCmdNameCommand = "command";
 const std::string kCmdNameDiskRecovery = "diskrecovery";
 const std::string kCmdNameClearReplicationID = "clearreplicationid";
+const std::string kCmdNameDisableWal = "disablewal";
 
 // Migrate slot
 const std::string kCmdNameSlotsMgrtSlot = "slotsmgrtslot";

src/pika_admin.cc

@@ -2660,6 +2660,28 @@ void ClearReplicationIDCmd::Do(std::shared_ptr<Slot> slot) {
   res_.SetRes(CmdRes::kOk, "ReplicationID is cleared");
 }
 
+void DisableWalCmd::DoInitial() {
+  if (!CheckArg(argv_.size())) {
+    res_.SetRes(CmdRes::kWrongNum, kCmdNameDisableWal);
+    return;
+  }
+}
+
+void DisableWalCmd::Do(std::shared_ptr<Slot> slot) {
+  std::string option = argv_[1].data();
+  bool is_wal_disable = false;
+  if (option.compare("true") == 0) {
+    is_wal_disable = true;
+  } else if (option.compare("false") == 0) {
+    is_wal_disable = false;
+  } else {
+    res_.SetRes(CmdRes::kErrOther, "Invalid parameter");
+    return;
+  }
+  slot->db()->DisableWal(is_wal_disable);
+  res_.SetRes(CmdRes::kOk, "Wal options is changed");
+}
+
 #ifdef WITH_COMMAND_DOCS
 
 bool CommandCmd::CommandFieldCompare::operator()(const std::string& a, const std::string& b) const {

src/pika_command.cc

@@ -96,6 +96,8 @@ void InitCmdTable(CmdTable* cmd_table) {
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameDiskRecovery, std::move(diskrecoveryptr)));
   std::unique_ptr<Cmd> clearreplicationidptr = std::make_unique<ClearReplicationIDCmd>(kCmdNameClearReplicationID, 1, kCmdFlagsWrite | kCmdFlagsAdmin);
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameClearReplicationID, std::move(clearreplicationidptr)));
+  std::unique_ptr<Cmd> disablewalptr = std::make_unique<DisableWalCmd>(kCmdNameDisableWal, 2, kCmdFlagsAdmin);
+  cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameDisableWal, std::move(disablewalptr)));
 
 #ifdef WITH_COMMAND_DOCS
   std::unique_ptr<Cmd> commandptr = std::make_unique<CommandCmd>(kCmdNameCommand, -1, kCmdFlagsRead | kCmdFlagsAdmin);

src/storage/include/storage/storage.h

@@ -976,6 +976,9 @@ class Storage {
 
   Status Keys(const DataType& data_type, const std::string& pattern, std::vector<std::string>* keys);
 
+  // Dynamic switch WAL
+  void DisableWal(const bool is_wal_disable);
+
   // Iterate through all the data in the database.
   void ScanDatabase(const DataType& type);
 

src/storage/src/redis.cc

@@ -165,4 +165,8 @@ void Redis::GetRocksDBInfo(std::string &info, const char *prefix) {
     info.append(string_stream.str());
 }
 
+void Redis::SetWriteWalOptions(const bool is_wal_disable) {
+  default_write_options_.disableWAL = is_wal_disable;
+}
+
 }  // namespace storage

src/storage/src/redis.h

@@ -31,6 +31,7 @@ class Redis {
   rocksdb::DB* GetDB() { return db_; }
 
   Status SetOptions(const OptionType& option_type, const std::unordered_map<std::string, std::string>& options);
+  void SetWriteWalOptions(const bool is_wal_disable);
 
   // Common Commands
   virtual Status Open(const StorageOptions& storage_options, const std::string& db_path) = 0;

src/storage/src/storage.cc

@@ -1761,4 +1761,12 @@ void Storage::GetRocksDBInfo(std::string& info) {
   zsets_db_->GetRocksDBInfo(info, "zsets_");
 }
 
+void Storage::DisableWal(const bool is_wal_disable) {
+  strings_db_->SetWriteWalOptions(is_wal_disable);
+  hashes_db_->SetWriteWalOptions(is_wal_disable);
+  lists_db_->SetWriteWalOptions(is_wal_disable);
+  sets_db_->SetWriteWalOptions(is_wal_disable);
+  zsets_db_->SetWriteWalOptions(is_wal_disable);
+}
+
 }  //  namespace storage