HrangebyLex supports specify intervals

src/commands/redis_cmd.cc

@@ -1508,25 +1508,60 @@ class CommandHGetAll : public Commander {
 class CommandHRange : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
-    if (args.size() != 6 && args.size() != 4) {
-      return Status(Status::RedisParseErr, errWrongNumOfArguments);
-    }
-    if (args.size() == 6 && Util::ToLower(args[4]) != "limit") {
-      return Status(Status::RedisInvalidCmd, errInvalidSyntax);
+    CommandParser parser(args, 4);
+    while (parser.Good()) {
+      if (parser.EatEqICaseFlag("BYLEX", by_flag_)) {
+      } else if (parser.EatEqICase("REV")) {
+        reversed_ = true;
+      } else if (parser.EatEqICase("LIMIT")) {
+        offset_ = GET_OR_RET(parser.TakeInt());
+        count_ = GET_OR_RET(parser.TakeInt());
+      } else {
+        return parser.InvalidSyntax();
+      }
     }
-    if (args.size() == 6) {
-      auto parse_result = ParseInt<int64_t>(args_[5], 10);
-      if (!parse_result) return Status(Status::RedisParseErr, errValueNotInteger);
-      limit_ = *parse_result;
+    if (by_flag_ == "BYLEX") {
+      spec_.reversed = reversed_;
+      spec_.count = count_;
+      spec_.offset = offset_;
+      Status s;
+      if (spec_.reversed) {
+        s = Redis::Hash::ParseRangeLexSpec(args[3], args[2], &spec_);
+      } else {
+        s = Redis::Hash::ParseRangeLexSpec(args[2], args[3], &spec_);
+      }
+      if (!s.IsOK()) {
+        return Status(Status::RedisParseErr, s.Msg());
+      }
+    } else {
+      by_flag_ = "BYINDEX";
+      auto parse_start = ParseInt<int>(args[2], 10);
+      auto parse_stop = ParseInt<int>(args[3], 10);
+      if (!parse_start || !parse_stop) {
+        return Status(Status::RedisParseErr, errValueNotInteger);
+      }
+      start_ = *parse_start;
+      stop_ = *parse_stop;
+      if (reversed_) std::swap(start_, stop_);
+      return Commander::Parse(args);
     }
-    return Commander::Parse(args);
+    return Status::OK();
   }
   Status Execute(Server *svr, Connection *conn, std::string *output) override {
     Redis::Hash hash_db(svr->storage_, conn->GetNamespace());
     std::vector<FieldValue> field_values;
-    rocksdb::Status s = hash_db.Range(args_[1], args_[2], args_[3], limit_, &field_values);
-    if (!s.ok()) {
-      return Status(Status::RedisExecErr, s.ToString());
+    if (by_flag_ == "BYLEX") {
+      rocksdb::Status s = hash_db.RangeByLex(args_[1], spec_, &field_values);
+      if (!s.ok()) {
+        return Status(Status::RedisExecErr, s.ToString());
+      }
+    } else if (by_flag_ == "BYINDEX") {
+      rocksdb::Status s = hash_db.Range(args_[1], start_, stop_, offset_, count_, reversed_, &field_values);
+      if (!s.ok()) {
+        return Status(Status::RedisExecErr, s.ToString());
+      }
+    } else {
+      assert(false);
     }
     std::vector<std::string> kv_pairs;
     for (const auto &p : field_values) {
@@ -1538,7 +1573,13 @@ class CommandHRange : public Commander {
   }
 
  private:
-  int64_t limit_ = LONG_MAX;
+  std::string_view by_flag_ = "";
+  bool reversed_ = false;
+  int64_t offset_ = 0;
+  int64_t count_ = -1;
+  HashSpec spec_;
+  int64_t start_;
+  int64_t stop_;
 };
 
 class CommandPush : public Commander {

src/types/redis_hash.cc

@@ -276,37 +276,121 @@ rocksdb::Status Hash::MSet(const Slice &user_key, const std::vector<FieldValue>
   return storage_->Write(storage_->DefaultWriteOptions(), &batch);
 }
 
-rocksdb::Status Hash::Range(const Slice &user_key, const Slice &start, const Slice &stop, int64_t limit,
-                            std::vector<FieldValue> *field_values) {
+rocksdb::Status Hash::RangeByLex(const Slice &user_key, const HashSpec &spec, std::vector<FieldValue> *field_values) {
   field_values->clear();
-  if (start.compare(stop) >= 0 || limit <= 0) {
+  if (spec.count == 0) {
     return rocksdb::Status::OK();
   }
   std::string ns_key;
   AppendNamespacePrefix(user_key, &ns_key);
   HashMetadata metadata(false);
   rocksdb::Status s = GetMetadata(ns_key, &metadata);
   if (!s.ok()) return s.IsNotFound() ? rocksdb::Status::OK() : s;
-  limit = std::min(static_cast<int64_t>(metadata.size), limit);
-  std::string start_key, stop_key;
-  InternalKey(ns_key, start, metadata.version, storage_->IsSlotIdEncoded()).Encode(&start_key);
-  InternalKey(ns_key, stop, metadata.version, storage_->IsSlotIdEncoded()).Encode(&stop_key);
+
+  std::string start_member = spec.reversed ? spec.max : spec.min;
+  std::string start_key, prefix_key, next_version_prefix_key;
+  InternalKey(ns_key, start_member, metadata.version, storage_->IsSlotIdEncoded()).Encode(&start_key);
+  InternalKey(ns_key, "", metadata.version, storage_->IsSlotIdEncoded()).Encode(&prefix_key);
+  InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode(&next_version_prefix_key);
   rocksdb::ReadOptions read_options;
   LatestSnapShot ss(db_);
   read_options.snapshot = ss.GetSnapShot();
-  rocksdb::Slice upper_bound(stop_key);
+  rocksdb::Slice upper_bound(next_version_prefix_key);
   read_options.iterate_upper_bound = &upper_bound;
+  rocksdb::Slice lower_bound(prefix_key);
+  read_options.iterate_lower_bound = &lower_bound;
   read_options.fill_cache = false;
 
   auto iter = DBUtil::UniqueIterator(db_, read_options);
-  iter->Seek(start_key);
-  for (int64_t i = 0; iter->Valid() && i <= limit - 1; ++i) {
+  if (!spec.reversed) {
+    iter->Seek(start_key);
+  } else {
+    if (spec.max_infinite) {
+      iter->SeekToLast();
+    } else {
+      iter->SeekForPrev(start_key);
+    }
+  }
+  int64_t pos = 0;
+  for (; iter->Valid() && iter->key().starts_with(prefix_key); (!spec.reversed ? iter->Next() : iter->Prev())) {
     FieldValue tmp_field_value;
     InternalKey ikey(iter->key(), storage_->IsSlotIdEncoded());
+    if (spec.reversed) {
+      if (ikey.GetSubKey().ToString() < spec.min || (spec.minex && ikey.GetSubKey().ToString() == spec.min)) {
+        break;
+      }
+      if ((spec.maxex && ikey.GetSubKey().ToString() == spec.max) ||
+          (!spec.max_infinite && ikey.GetSubKey().ToString() > spec.max)) {
+        continue;
+      }
+    } else {
+      if (spec.minex && ikey.GetSubKey().ToString() == spec.min) continue;  // the min member was exclusive
+      if ((spec.maxex && ikey.GetSubKey().ToString() == spec.max) ||
+          (!spec.max_infinite && ikey.GetSubKey().ToString() > spec.max))
+        break;
+    }
+    if (spec.offset >= 0 && pos++ < spec.offset) continue;
     tmp_field_value.field = ikey.GetSubKey().ToString();
     tmp_field_value.value = iter->value().ToString();
     field_values->emplace_back(tmp_field_value);
-    iter->Next();
+    if (spec.count > 0 && field_values && field_values->size() >= static_cast<unsigned>(spec.count)) break;
+  }
+  return rocksdb::Status::OK();
+}
+rocksdb::Status Hash::Range(const Slice &user_key, int64_t start, int64_t stop, int64_t offset, int64_t count,
+                            bool reversed, std::vector<FieldValue> *field_values) {
+  field_values->clear();
+  if (count == 0) {
+    return rocksdb::Status::OK();
+  }
+  std::string ns_key;
+  AppendNamespacePrefix(user_key, &ns_key);
+  HashMetadata metadata(false);
+  rocksdb::Status s = GetMetadata(ns_key, &metadata);
+  if (!s.ok()) return s.IsNotFound() ? rocksdb::Status::OK() : s;
+  if (start < 0) start = static_cast<int64_t>(metadata.size) + start;
+  if (stop < 0) stop = static_cast<int64_t>(metadata.size) + stop;
+  if (start >= static_cast<int64_t>(metadata.size) || stop < 0 || start > stop) return rocksdb::Status::OK();
+  if (start < 0) start = 0;
+  if (count < 0) count = static_cast<int64_t>(metadata.size);
+  std::string prefix_key, next_version_prefix_key;
+  InternalKey(ns_key, "", metadata.version, storage_->IsSlotIdEncoded()).Encode(&prefix_key);
+  InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode(&next_version_prefix_key);
+  rocksdb::ReadOptions read_options;
+  LatestSnapShot ss(db_);
+  read_options.snapshot = ss.GetSnapShot();
+  rocksdb::Slice upper_bound(next_version_prefix_key);
+  read_options.iterate_upper_bound = &upper_bound;
+  read_options.fill_cache = false;
+
+  auto iter = DBUtil::UniqueIterator(db_, read_options);
+  iter->Seek(prefix_key);
+  if (reversed && (!iter->Valid() || !iter->key().starts_with(prefix_key))) {
+    iter->SeekForPrev(prefix_key);
+  }
+  if (!reversed) {
+    iter->Seek(prefix_key);
+  } else {
+    iter->SeekForPrev(next_version_prefix_key);
+  }
+  int64_t pos = 0;
+  // go to offset end
+  for (; iter->Valid() && iter->key().starts_with(prefix_key); !reversed ? iter->Next() : iter->Prev()) {
+    if (offset >= 0 && pos++ < offset)
+      continue;
+    else
+      break;
+  }
+  pos = 0;
+  count = std::min(count, stop - start + 1);
+  for (; iter->Valid() && iter->key().starts_with(prefix_key); !reversed ? iter->Next() : iter->Prev()) {
+    if (start >= 0 && pos++ < start) continue;
+    FieldValue tmp_field_value;
+    InternalKey ikey(iter->key(), storage_->IsSlotIdEncoded());
+    tmp_field_value.field = ikey.GetSubKey().ToString();
+    tmp_field_value.value = iter->value().ToString();
+    field_values->emplace_back(tmp_field_value);
+    if (count > 0 && field_values && field_values->size() >= static_cast<unsigned>(count)) break;
   }
   return rocksdb::Status::OK();
 }
@@ -355,4 +439,36 @@ rocksdb::Status Hash::Scan(const Slice &user_key, const std::string &cursor, uin
   return SubKeyScanner::Scan(kRedisHash, user_key, cursor, limit, field_prefix, fields, values);
 }
 
+Status Hash::ParseRangeLexSpec(const std::string &min, const std::string &max, HashSpec *spec) {
+  if (min == "+" || max == "-") {
+    return Status(Status::NotOK, "min > max");
+  }
+
+  if (min == "-") {
+    spec->min = "";
+  } else {
+    if (min[0] == '(') {
+      spec->minex = true;
+    } else if (min[0] == '[') {
+      spec->minex = false;
+    } else {
+      return Status(Status::NotOK, "the min is illegal");
+    }
+    spec->min = min.substr(1);
+  }
+
+  if (max == "+") {
+    spec->max_infinite = true;
+  } else {
+    if (max[0] == '(') {
+      spec->maxex = true;
+    } else if (max[0] == '[') {
+      spec->maxex = false;
+    } else {
+      return Status(Status::NotOK, "the max is illegal");
+    }
+    spec->max = max.substr(1);
+  }
+  return Status::OK();
+}
 }  // namespace Redis

src/types/redis_hash.h

@@ -29,13 +29,23 @@
 #include "storage/redis_db.h"
 #include "storage/redis_metadata.h"
 
-typedef struct FieldValue {
+struct FieldValue {
   std::string field;
   std::string value;
-} FieldValue;
+};
 
 enum class HashFetchType { kAll = 0, kOnlyKey = 1, kOnlyValue = 2 };
 
+struct HashSpec {
+  std::string min, max;
+  bool minex, maxex; /* are min or max exclusive */
+  bool max_infinite; /* are max infinite */
+  int64_t offset, count;
+  bool removed, reversed;
+  HashSpec()
+      : minex(false), maxex(false), max_infinite(false), offset(-1), count(-1), removed(false), reversed(false) {}
+};
+
 namespace Redis {
 class Hash : public SubKeyScanner {
  public:
@@ -48,8 +58,10 @@ class Hash : public SubKeyScanner {
   rocksdb::Status IncrBy(const Slice &user_key, const Slice &field, int64_t increment, int64_t *ret);
   rocksdb::Status IncrByFloat(const Slice &user_key, const Slice &field, double increment, double *ret);
   rocksdb::Status MSet(const Slice &user_key, const std::vector<FieldValue> &field_values, bool nx, int *ret);
-  rocksdb::Status Range(const Slice &user_key, const Slice &start, const Slice &stop, int64_t limit,
-                        std::vector<FieldValue> *field_values);
+  static Status ParseRangeLexSpec(const std::string &min, const std::string &max, HashSpec *spec);
+  rocksdb::Status RangeByLex(const Slice &user_key, const HashSpec &spec, std::vector<FieldValue> *field_values);
+  rocksdb::Status Range(const Slice &user_key, int64_t start, int64_t stop, int64_t offset, int64_t count,
+                        bool reversed, std::vector<FieldValue> *field_values);
   rocksdb::Status MGet(const Slice &user_key, const std::vector<Slice> &fields, std::vector<std::string> *values,
                        std::vector<rocksdb::Status> *statuses);
   rocksdb::Status GetAll(const Slice &user_key, std::vector<FieldValue> *field_values,