fix: some streams errors such as pkpatternmatchdel etc (OpenAtomFoundation#2726)

* fix pkpatternmatchdel error

---------

Co-authored-by: wangshaoyi <wangshaoyi@360.cn>

Author: wangshao1

src/storage/src/base_filter.h

@@ -16,6 +16,7 @@
 #include "src/base_value_format.h"
 #include "src/base_meta_value_format.h"
 #include "src/lists_meta_value_format.h"
+#include "src/pika_stream_meta_value.h"
 #include "src/strings_value_format.h"
 #include "src/zsets_data_key_format.h"
 #include "src/debug.h"
@@ -36,11 +37,12 @@ class BaseMetaFilter : public rocksdb::CompactionFilter {
      * The field designs of the remaining zset,set,hash and stream in meta-value
      * are the same, so the same filtering strategy is used
      */
+    ParsedBaseKey parsed_key(key);
     auto type = static_cast<enum DataType>(static_cast<uint8_t>(value[0]));
     DEBUG("==========================START==========================");
     if (type == DataType::kStrings) {
       ParsedStringsValue parsed_strings_value(value);
-      DEBUG("[StringsFilter]  key: {}, value = {}, timestamp: {}, cur_time: {}", key.ToString().c_str(),
+      DEBUG("[string type]  key: %s, value = %s, timestamp: %llu, cur_time: %llu", parsed_key.Key().ToString().c_str(),
             parsed_strings_value.UserValue().ToString().c_str(), parsed_strings_value.Etime(), cur_time);
       if (parsed_strings_value.Etime() != 0 && parsed_strings_value.Etime() < cur_time) {
         DEBUG("Drop[Stale]");
@@ -49,9 +51,17 @@ class BaseMetaFilter : public rocksdb::CompactionFilter {
         DEBUG("Reserve");
         return false;
       }
+    } else if (type == DataType::kStreams) {
+      ParsedStreamMetaValue parsed_stream_meta_value(value);
+      DEBUG("[stream meta type], key: %s, entries_added = %llu, first_id: %s, last_id: %s, version: %llu",
+            parsed_key.Key().ToString().c_str(), parsed_stream_meta_value.entries_added(),
+            parsed_stream_meta_value.first_id().ToString().c_str(),
+            parsed_stream_meta_value.last_id().ToString().c_str(),
+            parsed_stream_meta_value.version());
+      return false;
     } else if (type == DataType::kLists) {
       ParsedListsMetaValue parsed_lists_meta_value(value);
-      DEBUG("[ListMetaFilter], key: {}, count = {}, timestamp: {}, cur_time: {}, version: {}", key.ToString().c_str(),
+      DEBUG("[list meta type], key: %s, count = %d, timestamp: %llu, cur_time: %llu, version: %llu", parsed_key.Key().ToString().c_str(),
             parsed_lists_meta_value.Count(), parsed_lists_meta_value.Etime(), cur_time,
             parsed_lists_meta_value.Version());
 
@@ -68,8 +78,9 @@ class BaseMetaFilter : public rocksdb::CompactionFilter {
       return false;
     } else {
       ParsedBaseMetaValue parsed_base_meta_value(value);
-      DEBUG("[MetaFilter]  key: {}, count = {}, timestamp: {}, cur_time: {}, version: {}", key.ToString().c_str(),
-            parsed_base_meta_value.Count(), parsed_base_meta_value.Etime(), cur_time, parsed_base_meta_value.Version());
+      DEBUG("[%s meta type]  key: %s, count = %d, timestamp: %llu, cur_time: %llu, version: %llu",
+            DataTypeToString(type), parsed_key.Key().ToString().c_str(), parsed_base_meta_value.Count(),
+            parsed_base_meta_value.Etime(), cur_time, parsed_base_meta_value.Version());
 
       if (parsed_base_meta_value.Etime() != 0 && parsed_base_meta_value.Etime() < cur_time &&
           parsed_base_meta_value.Version() < cur_time) {
@@ -143,7 +154,12 @@ class BaseDataFilter : public rocksdb::CompactionFilter {
         auto type = static_cast<enum DataType>(static_cast<uint8_t>(meta_value[0]));
         if (type != type_) {
           return true;
-        } else if (type == DataType::kHashes || type == DataType::kSets || type == DataType::kStreams || type == DataType::kZSets) {
+        } else if (type == DataType::kStreams) {
+          ParsedStreamMetaValue parsed_stream_meta_value(meta_value);
+          meta_not_found_ = false;
+          cur_meta_version_ = parsed_stream_meta_value.version();
+          cur_meta_etime_ = 0; // stream do not support ttl
+        } else if (type == DataType::kHashes || type == DataType::kSets || type == DataType::kZSets) {
           ParsedBaseMetaValue parsed_base_meta_value(&meta_value);
           meta_not_found_ = false;
           cur_meta_version_ = parsed_base_meta_value.Version();

src/storage/src/pika_stream_meta_value.h

@@ -82,7 +82,8 @@ class StreamMetaValue {
     value_ = std::move(value);
     assert(value_.size() == kDefaultStreamValueLength);
     if (value_.size() != kDefaultStreamValueLength) {
-      LOG(ERROR) << "Invalid stream meta value length: ";
+      LOG(ERROR) << "Invalid stream meta value length: " << value_.size()
+                 << " expected: " << kDefaultStreamValueLength;
       return;
     }
     char* pos = &value_[0];
@@ -215,7 +216,8 @@ class ParsedStreamMetaValue {
   ParsedStreamMetaValue(const Slice& value) {
     assert(value.size() == kDefaultStreamValueLength);
     if (value.size() != kDefaultStreamValueLength) {
-      LOG(ERROR) << "Invalid stream meta value length: ";
+      LOG(ERROR) << "Invalid stream meta value length: " << value.size()
+                 << " expected: " << kDefaultStreamValueLength;
       return;
     }
     char* pos = const_cast<char*>(value.data());
@@ -294,7 +296,7 @@ class StreamCGroupMetaValue {
     uint64_t needed = kDefaultStreamCGroupValueLength;
     assert(value_.size() == 0);
     if (value_.size() != 0) {
-      LOG(FATAL) << "Init on a existed stream cgroup meta value!";
+      LOG(ERROR) << "Init on a existed stream cgroup meta value!";
       return;
     }
     value_.resize(needed);
@@ -314,7 +316,8 @@ class StreamCGroupMetaValue {
     value_ = std::move(value);
     assert(value_.size() == kDefaultStreamCGroupValueLength);
     if (value_.size() != kDefaultStreamCGroupValueLength) {
-      LOG(FATAL) << "Invalid stream cgroup meta value length: ";
+      LOG(ERROR) << "Invalid stream cgroup meta value length: " << value_.size()
+                 << " expected: " << kDefaultStreamValueLength;
       return;
     }
     if (value_.size() == kDefaultStreamCGroupValueLength) {
@@ -373,7 +376,7 @@ class StreamConsumerMetaValue {
     value_ = std::move(value);
     assert(value_.size() == kDefaultStreamConsumerValueLength);
     if (value_.size() != kDefaultStreamConsumerValueLength) {
-      LOG(FATAL) << "Invalid stream consumer meta value length: " << value_.size()
+      LOG(ERROR) << "Invalid stream consumer meta value length: " << value_.size()
                  << " expected: " << kDefaultStreamConsumerValueLength;
       return;
     }
@@ -391,7 +394,7 @@ class StreamConsumerMetaValue {
     pel_ = pel;
     assert(value_.size() == 0);
     if (value_.size() != 0) {
-      LOG(FATAL) << "Invalid stream consumer meta value length: " << value_.size() << " expected: 0";
+      LOG(ERROR) << "Invalid stream consumer meta value length: " << value_.size() << " expected: 0";
       return;
     }
     uint64_t needed = kDefaultStreamConsumerValueLength;

src/storage/src/redis_strings.cc

@@ -1703,19 +1703,19 @@ rocksdb::Status Redis::PKPatternMatchDel(const std::string& pattern, int32_t* re
   rocksdb::WriteBatch batch;
   rocksdb::Iterator* iter = db_->NewIterator(iterator_options, handles_[kMetaCF]);
   iter->SeekToFirst();
-  key = iter->key().ToString();
   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) {
-      meta_value = iter->value().ToString();
       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);
       }
     } else if (meta_type == DataType::kLists) {
-      meta_value = iter->value().ToString();
       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) !=
@@ -1732,7 +1732,6 @@ rocksdb::Status Redis::PKPatternMatchDel(const std::string& pattern, int32_t* re
         batch.Put(handles_[kMetaCF], key, stream_meta_value.value());
       }
     } else {
-      meta_value = iter->value().ToString();
       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) !=

src/storage/src/storage.cc

@@ -1401,11 +1401,14 @@ Status Storage::PKRScanRange(const DataType& data_type, const Slice& key_start,
 
 Status Storage::PKPatternMatchDel(const DataType& data_type, const std::string& pattern, int32_t* ret) {
   Status s;
+  *ret = 0;
   for (const auto& inst : insts_) {
-    s = inst->PKPatternMatchDel(pattern, ret);
+    int32_t tmp_ret = 0;
+    s = inst->PKPatternMatchDel(pattern, &tmp_ret);
     if (!s.ok()) {
       return s;
     }
+    *ret += tmp_ret;
   }
   return s;
 }