fix:pkpatternmatchdel support stream type (OpenAtomFoundation#2723)

* pkmatchpatterndel support stream type * fix typo --------- Co-authored-by: wangshaoyi <[email protected]>
cheniujh · Jun 14, 2024 · 8c9b4d0 · 8c9b4d0
1 parent 5e02c72
commit 8c9b4d0
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 11 deletions.
diff --git a/src/pika_admin.cc b/src/pika_admin.cc
@@ -1210,9 +1210,9 @@ void InfoCmd::InfoKeyspace(std::string& info) {
   if (argv_.size() > 1 && strcasecmp(argv_[1].data(), kAllSection.data()) == 0) {
     tmp_stream << "# Start async statistics\r\n";
   } else if (argv_.size() == 3 && strcasecmp(argv_[1].data(), kKeyspaceSection.data()) == 0) {
-    tmp_stream << "# Start async statistics\r\n";    
+    tmp_stream << "# Start async statistics\r\n";
   } else {
-    tmp_stream << "# Use \"info keyspace 1\" to do async statistics\r\n";  
+    tmp_stream << "# Use \"info keyspace 1\" to do async statistics\r\n";
   }
   std::shared_lock rwl(g_pika_server->dbs_rw_);
   for (const auto& db_item : g_pika_server->dbs_) {
@@ -1236,7 +1236,7 @@ void InfoCmd::InfoKeyspace(std::string& info) {
         tmp_stream << "# Duration: " << std::to_string(duration) + "s"
                    << "\r\n";
       }
-      
+
       tmp_stream << db_name << " Strings_keys=" << key_infos[0].keys << ", expires=" << key_infos[0].expires
                  << ", invalid_keys=" << key_infos[0].invaild_keys << "\r\n";
       tmp_stream << db_name << " Hashes_keys=" << key_infos[1].keys << ", expires=" << key_infos[1].expires
@@ -2301,7 +2301,7 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     if (value != "true" && value != "false") {
       res_.AppendStringRaw("-ERR invalid disable_auto_compactions (true or false)\r\n");
       return;
-    } 
+    }
     std::unordered_map<std::string, std::string> options_map{{"disable_auto_compactions", value}};
     storage::Status s = g_pika_server->RewriteStorageOptions(storage::OptionType::kColumnFamily, options_map);
     if (!s.ok()) {
@@ -2929,6 +2929,8 @@ void PKPatternMatchDelCmd::DoInitial() {
     type_ = storage::kZSets;
   } else if (strcasecmp(argv_[2].data(), "hash") == 0) {
     type_ = storage::kHashes;
+  } else if (strcasecmp(argv_[2].data(), "stream") == 0) {
+    type_ = storage::kStreams;
   } else {
     res_.SetRes(CmdRes::kInvalidDbType, kCmdNamePKPatternMatchDel);
     return;

diff --git a/src/storage/include/storage/storage.h b/src/storage/include/storage/storage.h
@@ -1089,7 +1089,7 @@ class Storage {
   Status SetOptions(const OptionType& option_type, const std::string& db_type,
                     const std::unordered_map<std::string, std::string>& options);
   void SetCompactRangeOptions(const bool is_canceled);
-  Status EnableDymayticOptions(const OptionType& option_type, 
+  Status EnableDynamicOptions(const OptionType& option_type, 
                     const std::string& db_type, const std::unordered_map<std::string, std::string>& options);
   Status EnableAutoCompaction(const OptionType& option_type, 
                     const std::string& db_type, const std::unordered_map<std::string, std::string>& options);

diff --git a/src/storage/src/storage.cc b/src/storage/src/storage.cc
@@ -1155,6 +1155,9 @@ Status Storage::PKPatternMatchDel(const DataType& data_type, const std::string&
     case DataType::kSets:
       s = sets_db_->PKPatternMatchDel(pattern, ret);
       break;
+    case DataType::kStreams:
+      s = streams_db_->PKPatternMatchDel(pattern, ret);
+      break;
     default:
       s = Status::Corruption("Unsupported data type");
       break;
@@ -1921,7 +1924,7 @@ Status Storage::SetOptions(const OptionType& option_type, const std::string& db_
       return s;
     }
   }
-  s = EnableDymayticOptions(option_type,db_type,options);
+  s = EnableDynamicOptions(option_type,db_type,options);
   return s;
 }
 
@@ -1933,19 +1936,19 @@ void Storage::SetCompactRangeOptions(const bool is_canceled) {
   zsets_db_->SetCompactRangeOptions(is_canceled);
 }
 
-Status Storage::EnableDymayticOptions(const OptionType& option_type, 
+Status Storage::EnableDynamicOptions(const OptionType& option_type,
                             const std::string& db_type, const std::unordered_map<std::string, std::string>& options) {
   Status s;
   auto it = options.find("disable_auto_compactions");
   if (it != options.end() && it->second == "false") {
     s = EnableAutoCompaction(option_type,db_type,options);
-    LOG(WARNING) << "EnableAutoCompaction " << (s.ok() ? "success" : "failed") 
+    LOG(WARNING) << "EnableAutoCompaction " << (s.ok() ? "success" : "failed")
                  << " when Options get disable_auto_compactions: " << it->second << ",db_type:" << db_type;
   }
   return s;
 }
 
-Status Storage::EnableAutoCompaction(const OptionType& option_type, 
+Status Storage::EnableAutoCompaction(const OptionType& option_type,
                             const std::string& db_type, const std::unordered_map<std::string, std::string>& options){
   Status s;
   std::vector<std::string> cfs;
@@ -2030,8 +2033,8 @@ int64_t Storage::IsExist(const Slice& key, std::map<DataType, Status>* type_stat
   }
   return type_count;
 }
-  
-  
+
+
 void Storage::DisableWal(const bool is_wal_disable) {
   strings_db_->SetWriteWalOptions(is_wal_disable);
   hashes_db_->SetWriteWalOptions(is_wal_disable);