fix: PkPatternMatchDel inconsistent between rediscache and db

include/pika_admin.h

@@ -466,11 +466,15 @@ class PKPatternMatchDelCmd : public Cmd {
   PKPatternMatchDelCmd(const std::string& name, int arity, uint32_t flag)
       : Cmd(name, arity, flag, static_cast<uint32_t>(AclCategory::ADMIN)) {}
   void Do() override;
+  void DoThroughDB() override;
+  void DoUpdateCache() override;
   void Split(const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new PKPatternMatchDelCmd(*this); }
+  void DoBinlog() override;
 
  private:
+  std::vector<std::string> remove_keys_;
   storage::DataType type_ = storage::DataType::kAll;
   std::string pattern_;
   void DoInitial() override;

src/pika_admin.cc

@@ -3129,14 +3129,46 @@ void PKPatternMatchDelCmd::DoInitial() {
   pattern_ = argv_[1];
 }
 
-//TODO: may lead to inconsistent between rediscache and db, because currently it only cleans db
 void PKPatternMatchDelCmd::Do() {
-  int ret = 0;
-  rocksdb::Status s = db_->storage()->PKPatternMatchDel(type_, pattern_, &ret);
-  if (s.ok()) {
-    res_.AppendInteger(ret);
+  int64_t count = 0;
+  rocksdb::Status s = db_->storage()->PKPatternMatchDelWithRemoveKeys(type_, pattern_, &count, &remove_keys_);
+
+  if(s.ok()){
+    res_.AppendInteger(count);
+    s_ = rocksdb::Status::OK();
+    std::vector<std::string>::const_iterator it;
+    for (it = remove_keys_.begin(); it != remove_keys_.end(); it++) {
+      RemSlotKey(*it, db_);
+    }
   } else {
     res_.SetRes(CmdRes::kErrOther, s.ToString());
+    if (count >= 0) {
+      s_ = rocksdb::Status::OK();
+      std::vector<std::string>::const_iterator it;
+      for (it = remove_keys_.begin(); it != remove_keys_.end(); it++) {
+        RemSlotKey(*it, db_);
+      }
+    }
+  }
+}
+
+void PKPatternMatchDelCmd::DoThroughDB() {
+  Do();
+}
+
+void PKPatternMatchDelCmd::DoUpdateCache() {
+  if(s_.ok()) {
+    db_->cache()->Del(remove_keys_);
+  }
+}
+
+void PKPatternMatchDelCmd::DoBinlog() {
+  std::string opt = "del";
+  for(auto& key: remove_keys_) {
+    argv_.clear();
+    argv_.emplace_back(opt);
+    argv_.emplace_back(key);
+    Cmd::DoBinlog();
   }
 }
 

src/storage/include/storage/storage.h

@@ -987,6 +987,8 @@ class Storage {
   // the pattern
   Status PKPatternMatchDel(const DataType& data_type, const std::string& pattern, int32_t* ret);
 
+  Status PKPatternMatchDelWithRemoveKeys(const DataType& data_type, const std::string& pattern, int64_t* ret, std::vector<std::string>* remove_keys);
+
   // Iterate over a collection of elements
   // return next_key that the user need to use as the start_key argument
   // in the next call

src/storage/src/redis.h

@@ -190,6 +190,7 @@ class Redis {
   Status Persist(const Slice& key);
   Status TTL(const Slice& key, int64_t* timestamp);
   Status PKPatternMatchDel(const std::string& pattern, int32_t* ret);
+  Status PKPatternMatchDelWithRemoveKeys(const std::string& pattern, int64_t* ret, std::vector<std::string>* remove_keys);
 
   Status GetType(const Slice& key, enum DataType& type);
   Status IsExist(const Slice& key);

src/storage/src/redis_strings.cc

@@ -1748,6 +1748,7 @@ rocksdb::Status Redis::PKPatternMatchDel(const std::string& pattern, int32_t* re
         batch.Clear();
       } else {
         *ret = total_delete;
+        delete iter;
         return s;
       }
     }
@@ -1762,6 +1763,93 @@ rocksdb::Status Redis::PKPatternMatchDel(const std::string& pattern, int32_t* re
   }
 
   *ret = total_delete;
+  delete iter;
   return s;
 }
+
+rocksdb::Status Redis::PKPatternMatchDelWithRemoveKeys(const std::string& pattern, int64_t* ret, std::vector<std::string>* remove_keys) {
+  rocksdb::ReadOptions iterator_options;
+  const rocksdb::Snapshot* snapshot;
+  ScopeSnapshot ss(db_, &snapshot);
+  iterator_options.snapshot = snapshot;
+  iterator_options.fill_cache = false;
+
+  std::string key;
+  std::string meta_value;
+  int32_t total_delete = 0;
+  rocksdb::Status s;
+  rocksdb::WriteBatch batch;
+  rocksdb::Iterator* iter = db_->NewIterator(iterator_options, handles_[kMetaCF]);
+  iter->SeekToFirst();
+  while (iter->Valid()) {
+    auto meta_type = static_cast<enum DataType>(static_cast<uint8_t>(iter->value()[0]));
+    ParsedBaseMetaKey parsed_meta_key(iter->key().ToString());
+    key = iter->key().ToString();
+    meta_value = iter->value().ToString();
+
+    if (meta_type == DataType::kStrings) {
+      ParsedStringsValue parsed_strings_value(&meta_value);
+      if (!parsed_strings_value.IsStale() &&
+          (StringMatch(pattern.data(), pattern.size(), parsed_meta_key.Key().data(), parsed_meta_key.Key().size(), 0) != 0)) {
+        batch.Delete(key);
+        remove_keys->push_back(parsed_meta_key.Key().data());
+      }
+    } else if (meta_type == DataType::kLists) {
+      ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+      if (!parsed_lists_meta_value.IsStale() && (parsed_lists_meta_value.Count() != 0U) &&
+          (StringMatch(pattern.data(), pattern.size(), parsed_meta_key.Key().data(), parsed_meta_key.Key().size(), 0) !=
+           0)) {
+        parsed_lists_meta_value.InitialMetaValue();
+        batch.Put(handles_[kMetaCF], iter->key(), meta_value);
+        remove_keys->push_back(parsed_meta_key.Key().data());
+      }
+    } else if (meta_type == DataType::kStreams) {
+      StreamMetaValue stream_meta_value;
+      stream_meta_value.ParseFrom(meta_value);
+      if ((stream_meta_value.length() != 0) &&
+          (StringMatch(pattern.data(), pattern.size(), parsed_meta_key.Key().data(), parsed_meta_key.Key().size(), 0) != 0)) {
+        stream_meta_value.InitMetaValue();
+        batch.Put(handles_[kMetaCF], key, stream_meta_value.value());
+        remove_keys->push_back(parsed_meta_key.Key().data());
+      }
+    } else {
+      ParsedBaseMetaValue parsed_meta_value(&meta_value);
+      if (!parsed_meta_value.IsStale() && (parsed_meta_value.Count() != 0) &&
+          (StringMatch(pattern.data(), pattern.size(), parsed_meta_key.Key().data(), parsed_meta_key.Key().size(), 0) !=
+           0)) {
+        parsed_meta_value.InitialMetaValue();
+        batch.Put(handles_[kMetaCF], iter->key(), meta_value);
+        remove_keys->push_back(parsed_meta_key.Key().data());
+      }
+    }
+
+    if (static_cast<size_t>(batch.Count()) >= BATCH_DELETE_LIMIT) {
+      s = db_->Write(default_write_options_, &batch);
+      if (s.ok()) {
+        total_delete += static_cast<int32_t>(batch.Count());
+        batch.Clear();
+      } else {
+        remove_keys->erase(remove_keys->end() - batch.Count(), remove_keys->end());
+        delete iter;
+        *ret = total_delete;
+        return s;
+      }
+    }
+    iter->Next();
+  }
+  if (batch.Count() != 0U) {
+    s = db_->Write(default_write_options_, &batch);
+    if (s.ok()) {
+      total_delete += static_cast<int32_t>(batch.Count());
+      batch.Clear();
+    } else {
+      remove_keys->erase(remove_keys->end() - batch.Count(), remove_keys->end());
+    }
+  }
+
+  *ret = total_delete;
+  delete iter;
+  return s;
+}
+
 }  //  namespace storage

src/storage/src/storage.cc

@@ -1399,13 +1399,30 @@ Status Storage::PKRScanRange(const DataType& data_type, const Slice& key_start,
   return Status::OK();
 }
 
+Status Storage::PKPatternMatchDelWithRemoveKeys(const DataType& data_type, const std::string& pattern, 
+                                                 int64_t* ret, std::vector<std::string>* remove_keys) {
+  Status s;
+  *ret = 0;
+  for (const auto& inst : insts_) {
+    int64_t tmp_ret = 0;
+    s = inst->PKPatternMatchDelWithRemoveKeys(pattern, &tmp_ret, remove_keys);
+    if (!s.ok()) {
+      *ret += tmp_ret;
+      return s;
+    }
+    *ret += tmp_ret;
+  }
+  return s;
+}
+
 Status Storage::PKPatternMatchDel(const DataType& data_type, const std::string& pattern, int32_t* ret) {
   Status s;
   *ret = 0;
   for (const auto& inst : insts_) {
     int32_t tmp_ret = 0;
     s = inst->PKPatternMatchDel(pattern, &tmp_ret);
     if (!s.ok()) {
+      *ret += tmp_ret;
       return s;
     }
     *ret += tmp_ret;