diff --git a/src/server/hset_family.cc b/src/server/hset_family.cc
index 3d3586118cce..a8fcc870e694 100644
--- a/src/server/hset_family.cc
+++ b/src/server/hset_family.cc
@@ -27,6 +27,9 @@ using namespace facade;
 namespace {
 
 constexpr size_t kMaxListPackLen = 1024;
+using IncrByParam = std::variant<double, int64_t>;
+using OptStr = std::optional<std::string>;
+enum GetAllMode : uint8_t { FIELDS = 1, VALUES = 2 };
 
 bool IsGoodForListpack(CmdArgList args, const uint8_t* lp) {
   size_t sum = 0;
@@ -92,353 +95,425 @@ pair<uint8_t*, bool> LpInsert(uint8_t* lp, string_view field, string_view val, b
   return make_pair(lp, !updated);
 }
 
-}  // namespace
+OpStatus OpIncrBy(const OpArgs& op_args, string_view key, string_view field, IncrByParam* param) {
+  auto& db_slice = op_args.shard->db_slice();
+  const auto [it, inserted] = db_slice.AddOrFind(op_args.db_cntx, key);
 
-void HSetFamily::HDel(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
+  DbTableStats* stats = db_slice.MutableStats(op_args.db_cntx.db_index);
 
-  args.remove_prefix(2);
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpDel(t->GetOpArgs(shard), key, args);
-  };
+  robj* hset = nullptr;
+  size_t lpb = 0;
 
-  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result || result.status() == OpStatus::KEY_NOTFOUND) {
-    (*cntx)->SendLong(*result);
+  if (inserted) {
+    hset = createHashObject();
+    it->second.ImportRObj(hset);
+    stats->listpack_blob_cnt++;
+    hset = it->second.AsRObj();
   } else {
-    (*cntx)->SendError(result.status());
-  }
-}
+    if (it->second.ObjType() != OBJ_HASH)
+      return OpStatus::WRONG_TYPE;
 
-void HSetFamily::HLen(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
+    db_slice.PreUpdate(op_args.db_cntx.db_index, it);
+    hset = it->second.AsRObj();
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpLen(t->GetOpArgs(shard), key);
-  };
+    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+      lpb = lpBytes((uint8_t*)hset->ptr);
+      stats->listpack_bytes -= lpb;
 
-  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    (*cntx)->SendLong(*result);
-  } else {
-    (*cntx)->SendError(result.status());
+      if (lpb >= kMaxListPackLen) {
+        stats->listpack_blob_cnt--;
+        hashTypeConvert(hset, OBJ_ENCODING_HT);
+      }
+    }
   }
-}
 
-void HSetFamily::HExists(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  string_view field = ArgS(args, 2);
+  unsigned char* vstr = NULL;
+  unsigned int vlen = UINT_MAX;
+  long long old_val = 0;
 
-  auto cb = [&](Transaction* t, EngineShard* shard) -> OpResult<int> {
-    auto& db_slice = shard->db_slice();
-    auto it_res = db_slice.Find(t->db_context(), key, OBJ_HASH);
+  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
 
-    if (it_res) {
-      robj* hset = (*it_res)->second.AsRObj();
-      shard->tmp_str1 = sdscpylen(shard->tmp_str1, field.data(), field.size());
+  int exist_res = hashTypeGetValue(hset, op_args.shard->tmp_str1, &vstr, &vlen, &old_val);
 
-      return hashTypeExists(hset, shard->tmp_str1);
-    }
-    if (it_res.status() == OpStatus::KEY_NOTFOUND)
-      return 0;
-    return it_res.status();
-  };
+  if (holds_alternative<double>(*param)) {
+    long double value;
+    double incr = get<double>(*param);
+    if (exist_res == C_OK) {
+      if (vstr) {
+        const char* exist_val = reinterpret_cast<char*>(vstr);
 
-  OpResult<int> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    (*cntx)->SendLong(*result);
-  } else {
-    (*cntx)->SendError(result.status());
-  }
-}
+        if (!string2ld(exist_val, vlen, &value)) {
+          stats->listpack_bytes += lpb;
 
-void HSetFamily::HMGet(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
+          return OpStatus::INVALID_VALUE;
+        }
+      } else {
+        value = old_val;
+      }
+      value += incr;
 
-  args.remove_prefix(2);
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpMGet(t->GetOpArgs(shard), key, args);
-  };
+      if (isnan(value) || isinf(value)) {
+        return OpStatus::INVALID_FLOAT;
+      }
+    } else {
+      value = incr;
+    }
 
-  OpResult<vector<OptStr>> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+    char buf[128];
+    char* str = RedisReplyBuilder::FormatDouble(value, buf, sizeof(buf));
+    string_view sval{str};
 
-  if (result) {
-    (*cntx)->StartArray(result->size());
-    for (const auto& val : *result) {
-      if (val) {
-        (*cntx)->SendBulkString(*val);
-      } else {
-        (*cntx)->SendNull();
+    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+      uint8_t* lp = (uint8_t*)hset->ptr;
+
+      lp = LpInsert(lp, field, sval, false).first;
+      hset->ptr = lp;
+      stats->listpack_bytes += lpBytes(lp);
+    } else {
+      sds news = sdsnewlen(str, sval.size());
+      hashTypeSet(hset, op_args.shard->tmp_str1, news, HASH_SET_TAKE_VALUE);
+    }
+    param->emplace<double>(value);
+  } else {
+    if (exist_res == C_OK && vstr) {
+      const char* exist_val = reinterpret_cast<char*>(vstr);
+      if (!string2ll(exist_val, vlen, &old_val)) {
+        stats->listpack_bytes += lpb;
+
+        return OpStatus::INVALID_VALUE;
       }
     }
-  } else if (result.status() == OpStatus::KEY_NOTFOUND) {
-    (*cntx)->StartArray(args.size());
-    for (unsigned i = 0; i < args.size(); ++i) {
-      (*cntx)->SendNull();
+    int64_t incr = get<int64_t>(*param);
+    if ((incr < 0 && old_val < 0 && incr < (LLONG_MIN - old_val)) ||
+        (incr > 0 && old_val > 0 && incr > (LLONG_MAX - old_val))) {
+      stats->listpack_bytes += lpb;
+
+      return OpStatus::OUT_OF_RANGE;
     }
-  } else {
-    (*cntx)->SendError(result.status());
-  }
-}
 
-void HSetFamily::HGet(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  string_view field = ArgS(args, 2);
+    int64_t new_val = old_val + incr;
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpGet(t->GetOpArgs(shard), key, field);
-  };
+    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+      char buf[32];
+      char* next = absl::numbers_internal::FastIntToBuffer(new_val, buf);
+      string_view sval{buf, size_t(next - buf)};
+      uint8_t* lp = (uint8_t*)hset->ptr;
 
-  OpResult<string> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    (*cntx)->SendBulkString(*result);
-  } else {
-    if (result.status() == OpStatus::KEY_NOTFOUND) {
-      (*cntx)->SendNull();
+      lp = LpInsert(lp, field, sval, false).first;
+      hset->ptr = lp;
+      stats->listpack_bytes += lpBytes(lp);
     } else {
-      (*cntx)->SendError(result.status());
+      sds news = sdsfromlonglong(new_val);
+      hashTypeSet(hset, op_args.shard->tmp_str1, news, HASH_SET_TAKE_VALUE);
     }
+    param->emplace<int64_t>(new_val);
   }
+
+  it->second.SyncRObj();
+  db_slice.PostUpdate(op_args.db_cntx.db_index, it, key);
+
+  return OpStatus::OK;
 }
 
-void HSetFamily::HIncrBy(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  string_view field = ArgS(args, 2);
-  string_view incrs = ArgS(args, 3);
-  int64_t ival = 0;
+OpResult<StringVec> OpScan(const OpArgs& op_args, std::string_view key, uint64_t* cursor) {
+  OpResult<PrimeIterator> find_res = op_args.shard->db_slice().Find(op_args.db_cntx, key, OBJ_HASH);
 
-  if (!absl::SimpleAtoi(incrs, &ival)) {
-    return (*cntx)->SendError(kInvalidIntErr);
-  }
+  if (!find_res)
+    return find_res.status();
 
-  IncrByParam param{ival};
+  PrimeIterator it = find_res.value();
+  StringVec res;
+  uint32_t count = 20;
+  robj* hset = it->second.AsRObj();
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpIncrBy(t->GetOpArgs(shard), key, field, &param);
-  };
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    uint8_t* lp = (uint8_t*)hset->ptr;
+    uint8_t* lp_elem = lpFirst(lp);
 
-  OpStatus status = cntx->transaction->ScheduleSingleHop(std::move(cb));
+    DCHECK(lp_elem);  // empty containers are not allowed.
 
-  if (status == OpStatus::OK) {
-    (*cntx)->SendLong(get<int64_t>(param));
-  } else {
-    switch (status) {
-      case OpStatus::INVALID_VALUE:
-        (*cntx)->SendError("hash value is not an integer");
-        break;
-      case OpStatus::OUT_OF_RANGE:
-        (*cntx)->SendError(kIncrOverflow);
-        break;
-      default:
-        (*cntx)->SendError(status);
-        break;
-    }
-  }
-}
+    int64_t ele_len;
+    unsigned char intbuf[LP_INTBUF_SIZE];
 
-void HSetFamily::HIncrByFloat(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  string_view field = ArgS(args, 2);
-  string_view incrs = ArgS(args, 3);
-  double dval = 0;
+    // We do single pass on listpack for this operation.
+    do {
+      uint8_t* elem = lpGet(lp_elem, &ele_len, intbuf);
+      DCHECK(elem);
+      res.emplace_back(reinterpret_cast<char*>(elem), size_t(ele_len));
+      lp_elem = lpNext(lp, lp_elem);  // switch to value
+    } while (lp_elem);
+    *cursor = 0;
+  } else {
+    dict* ht = (dict*)hset->ptr;
+    long maxiterations = count * 10;
+    void* privdata = &res;
+    auto scanCb = [](void* privdata, const dictEntry* de) {
+      StringVec* res = (StringVec*)privdata;
+      sds val = (sds)de->key;
+      res->emplace_back(val, sdslen(val));
+      val = (sds)de->v.val;
+      res->emplace_back(val, sdslen(val));
+    };
 
-  if (!absl::SimpleAtod(incrs, &dval)) {
-    return (*cntx)->SendError(kInvalidFloatErr);
+    do {
+      *cursor = dictScan(ht, *cursor, scanCb, NULL, privdata);
+    } while (*cursor && maxiterations-- && res.size() < count);
   }
 
-  IncrByParam param{dval};
+  return res;
+}
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpIncrBy(t->GetOpArgs(shard), key, field, &param);
-  };
+OpResult<uint32_t> OpDel(const OpArgs& op_args, string_view key, CmdArgList values) {
+  DCHECK(!values.empty());
 
-  OpStatus status = cntx->transaction->ScheduleSingleHop(std::move(cb));
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
 
-  if (status == OpStatus::OK) {
-    (*cntx)->SendDouble(get<double>(param));
-  } else {
-    switch (status) {
-      case OpStatus::INVALID_VALUE:
-        (*cntx)->SendError("hash value is not a float");
-        break;
-      default:
-        (*cntx)->SendError(status);
+  if (!it_res)
+    return it_res.status();
+
+  db_slice.PreUpdate(op_args.db_cntx.db_index, *it_res);
+  CompactObj& co = (*it_res)->second;
+  robj* hset = co.AsRObj();
+  unsigned deleted = 0;
+  bool key_remove = false;
+  DbTableStats* stats = db_slice.MutableStats(op_args.db_cntx.db_index);
+
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    stats->listpack_bytes -= lpBytes((uint8_t*)hset->ptr);
+  }
+
+  for (auto s : values) {
+    op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, s.data(), s.size());
+
+    if (hashTypeDelete(hset, op_args.shard->tmp_str1)) {
+      ++deleted;
+      if (hashTypeLength(hset) == 0) {
+        key_remove = true;
         break;
+      }
     }
   }
-}
 
-void HSetFamily::HKeys(CmdArgList args, ConnectionContext* cntx) {
-  HGetGeneric(args, cntx, FIELDS);
-}
+  co.SyncRObj();
 
-void HSetFamily::HVals(CmdArgList args, ConnectionContext* cntx) {
-  HGetGeneric(args, cntx, VALUES);
-}
+  db_slice.PostUpdate(op_args.db_cntx.db_index, *it_res, key);
+  if (key_remove) {
+    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+      stats->listpack_blob_cnt--;
+    }
+    db_slice.Del(op_args.db_cntx.db_index, *it_res);
+  } else if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    stats->listpack_bytes += lpBytes((uint8_t*)hset->ptr);
+  }
 
-void HSetFamily::HGetAll(CmdArgList args, ConnectionContext* cntx) {
-  HGetGeneric(args, cntx, GetAllMode::FIELDS | GetAllMode::VALUES);
+  return deleted;
 }
 
-void HSetFamily::HGetGeneric(CmdArgList args, ConnectionContext* cntx, uint8_t getall_mask) {
-  string_view key = ArgS(args, 1);
+OpResult<vector<OptStr>> OpMGet(const OpArgs& op_args, std::string_view key, CmdArgList fields) {
+  DCHECK(!fields.empty());
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpGetAll(t->GetOpArgs(shard), key, getall_mask);
-  };
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
 
-  OpResult<vector<string>> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (!it_res)
+    return it_res.status();
 
-  if (result) {
-    (*cntx)->SendStringArr(absl::Span<const string>{*result});
-  } else {
-    (*cntx)->SendError(result.status());
-  }
-}
+  CompactObj& co = (*it_res)->second;
+  robj* hset = co.AsRObj();
 
-void HSetFamily::HScan(CmdArgList args, ConnectionContext* cntx) {
-  std::string_view key = ArgS(args, 1);
-  std::string_view token = ArgS(args, 2);
+  std::vector<OptStr> result(fields.size());
 
-  uint64_t cursor = 0;
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    uint8_t* lp = (uint8_t*)hset->ptr;
+    absl::flat_hash_map<string_view, unsigned> reverse;
+    reverse.reserve(fields.size() + 1);
+    for (size_t i = 0; i < fields.size(); ++i) {
+      reverse.emplace(ArgS(fields, i), i);  // map fields to their index.
+    }
 
-  if (!absl::SimpleAtoi(token, &cursor)) {
-    return (*cntx)->SendError("invalid cursor");
-  }
+    char ibuf[32];
+    uint8_t* lp_elem = lpFirst(lp);
+    int64_t ele_len;
+    string_view key;
+    DCHECK(lp_elem);  // empty containers are not allowed.
+    size_t lplen = lpLength(lp);
+    DCHECK(lplen > 0 && lplen % 2 == 0);
 
-  if (args.size() > 3) {
-    return (*cntx)->SendError("scan options are not supported yet");
-  }
+    // We do single pass on listpack for this operation.
+    do {
+      uint8_t* elem = lpGet(lp_elem, &ele_len, NULL);
+      if (elem) {
+        key = string_view{reinterpret_cast<char*>(elem), size_t(ele_len)};
+      } else {
+        char* next = absl::numbers_internal::FastIntToBuffer(ele_len, ibuf);
+        key = string_view{ibuf, size_t(next - ibuf)};
+      }
+      lp_elem = lpNext(lp, lp_elem);  // switch to value
+      DCHECK(lp_elem);
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpScan(t->GetOpArgs(shard), key, &cursor);
-  };
+      auto it = reverse.find(key);
+      if (it != reverse.end()) {
+        DCHECK_LT(it->second, result.size());
+        result[it->second].emplace(LpGetVal(lp_elem));  // populate found items.
+      }
 
-  OpResult<StringVec> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result.status() != OpStatus::WRONG_TYPE) {
-    (*cntx)->StartArray(2);
-    (*cntx)->SendSimpleString(absl::StrCat(cursor));
-    (*cntx)->StartArray(result->size());
-    for (const auto& k : *result) {
-      (*cntx)->SendBulkString(k);
-    }
+      lp_elem = lpNext(lp, lp_elem);  // switch to the next key
+    } while (lp_elem);
   } else {
-    (*cntx)->SendError(result.status());
+    DCHECK_EQ(OBJ_ENCODING_HT, hset->encoding);
+    dict* d = (dict*)hset->ptr;
+    for (size_t i = 0; i < fields.size(); ++i) {
+      op_args.shard->tmp_str1 =
+          sdscpylen(op_args.shard->tmp_str1, fields[i].data(), fields[i].size());
+      dictEntry* de = dictFind(d, op_args.shard->tmp_str1);
+      if (de) {
+        sds val = (sds)dictGetVal(de);
+        result[i].emplace(val, sdslen(val));
+      }
+    }
   }
-}
-
-void HSetFamily::HSet(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  ToLower(&args[0]);
-
-  string_view cmd = ArgS(args, 0);
 
-  if (args.size() % 2 != 0) {
-    return (*cntx)->SendError(facade::WrongNumArgsError(cmd), kSyntaxErrType);
-  }
+  return result;
+}
 
-  args.remove_prefix(2);
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpSet(t->GetOpArgs(shard), key, args, false);
-  };
+OpResult<uint32_t> OpLen(const OpArgs& op_args, string_view key) {
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
 
-  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result && cmd == "hset") {
-    (*cntx)->SendLong(*result);
-  } else {
-    (*cntx)->SendError(result.status());
+  if (it_res) {
+    robj* hset = (*it_res)->second.AsRObj();
+    return hashTypeLength(hset);
   }
+  if (it_res.status() == OpStatus::KEY_NOTFOUND)
+    return 0;
+  return it_res.status();
 }
 
-void HSetFamily::HSetNx(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
+OpResult<string> OpGet(const OpArgs& op_args, string_view key, string_view field) {
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+  if (!it_res)
+    return it_res.status();
 
-  args.remove_prefix(2);
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpSet(t->GetOpArgs(shard), key, args, true);
-  };
+  robj* hset = (*it_res)->second.AsRObj();
 
-  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    (*cntx)->SendLong(*result);
-  } else {
-    (*cntx)->SendError(result.status());
-  }
-}
+  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
 
-void HSetFamily::HStrLen(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
-  string_view field = ArgS(args, 2);
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    unsigned char* vstr = NULL;
+    unsigned int vlen = UINT_MAX;
+    long long vll = LLONG_MAX;
 
-  auto cb = [&](Transaction* t, EngineShard* shard) {
-    return OpStrLen(t->GetOpArgs(shard), key, field);
-  };
+    int ret = hashTypeGetFromListpack(hset, op_args.shard->tmp_str1, &vstr, &vlen, &vll);
+    if (ret < 0) {
+      return OpStatus::KEY_NOTFOUND;
+    }
+    if (vstr) {
+      const char* src = reinterpret_cast<const char*>(vstr);
+      return string{src, vlen};
+    }
 
-  OpResult<size_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    (*cntx)->SendLong(*result);
-  } else {
-    (*cntx)->SendError(result.status());
+    return absl::StrCat(vll);
   }
-}
+  DCHECK_EQ(hset->encoding, OBJ_ENCODING_HT);
 
-void HSetFamily::HRandField(CmdArgList args, ConnectionContext* cntx) {
-  string_view key = ArgS(args, 1);
+  dictEntry* de = dictFind((dict*)hset->ptr, op_args.shard->tmp_str1);
+  if (!de)
+    return OpStatus::KEY_NOTFOUND;
 
-  auto cb = [&](Transaction* t, EngineShard* shard) -> OpResult<StringVec> {
-    auto& db_slice = shard->db_slice();
-    auto it_res = db_slice.Find(t->db_context(), key, OBJ_HASH);
+  sds val = (sds)dictGetVal(de);
 
-    if (!it_res)
-      return it_res.status();
+  return string(val, sdslen(val));
+}
 
-    const PrimeValue& pv = it_res.value()->second;
-    StringVec str_vec;
+OpResult<vector<string>> OpGetAll(const OpArgs& op_args, string_view key, uint8_t mask) {
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+  if (!it_res) {
+    if (it_res.status() == OpStatus::KEY_NOTFOUND)
+      return vector<string>{};
+    return it_res.status();
+  }
 
-    if (pv.Encoding() == OBJ_ENCODING_HT) {
-      dict* this_dict = (dict*)pv.RObjPtr();
-      dictEntry* de = dictGetFairRandomKey(this_dict);
-      sds key = (sds)de->key;
-      str_vec.emplace_back(key, sdslen(key));
-    } else if (pv.Encoding() == OBJ_ENCODING_LISTPACK) {
-      uint8_t* lp = (uint8_t*)pv.RObjPtr();
-      size_t lplen = lpLength(lp);
-      CHECK(lplen > 0 && lplen % 2 == 0);
+  robj* hset = (*it_res)->second.AsRObj();
+  hashTypeIterator* hi = hashTypeInitIterator(hset);
 
-      size_t hlen = lplen / 2;
-      listpackEntry key;
+  vector<string> res;
+  bool keyval = (mask == (FIELDS | VALUES));
+  size_t len = hashTypeLength(hset);
+  res.resize(keyval ? len * 2 : len);
+  unsigned index = 0;
 
-      lpRandomPair(lp, hlen, &key, NULL);
-      if (key.sval) {
-        str_vec.emplace_back(reinterpret_cast<char*>(key.sval), key.slen);
-      } else {
-        str_vec.emplace_back(absl::StrCat(key.lval));
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    while (hashTypeNext(hi) != C_ERR) {
+      if (mask & FIELDS) {
+        res[index++] = LpGetVal(hi->fptr);
       }
-    } else {
-      LOG(ERROR) << "Invalid encoding " << pv.Encoding();
-      return OpStatus::INVALID_VALUE;
-    }
-    return str_vec;
-  };
 
-  OpResult<StringVec> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
-  if (result) {
-    CHECK_EQ(1u, result->size());  // TBD: to support count and withvalues.
-    (*cntx)->SendBulkString(result->front());
+      if (mask & VALUES) {
+        res[index++] = LpGetVal(hi->vptr);
+      }
+    }
   } else {
-    (*cntx)->SendError(result.status());
+    while (hashTypeNext(hi) != C_ERR) {
+      if (mask & FIELDS) {
+        sds key = (sds)dictGetKey(hi->de);
+        res[index++].assign(key, sdslen(key));
+      }
+
+      if (mask & VALUES) {
+        sds val = (sds)dictGetVal(hi->de);
+        res[index++].assign(val, sdslen(val));
+      }
+    }
+  }
+
+  hashTypeReleaseIterator(hi);
+
+  return res;
+}
+
+OpResult<size_t> OpStrLen(const OpArgs& op_args, string_view key, string_view field) {
+  auto& db_slice = op_args.shard->db_slice();
+  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+
+  if (!it_res) {
+    if (it_res.status() == OpStatus::KEY_NOTFOUND)
+      return 0;
+    return it_res.status();
+  }
+
+  robj* hset = (*it_res)->second.AsRObj();
+  size_t field_len = 0;
+  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
+
+  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
+    unsigned char* vstr = NULL;
+    unsigned int vlen = UINT_MAX;
+    long long vll = LLONG_MAX;
+
+    if (hashTypeGetFromListpack(hset, op_args.shard->tmp_str1, &vstr, &vlen, &vll) == 0)
+      field_len = vstr ? vlen : sdigits10(vll);
+
+    return field_len;
   }
+
+  DCHECK_EQ(hset->encoding, OBJ_ENCODING_HT);
+
+  dictEntry* de = dictFind((dict*)hset->ptr, op_args.shard->tmp_str1);
+  return de ? sdslen((sds)de->v.val) : 0;
 }
 
-OpResult<uint32_t> HSetFamily::OpSet(const OpArgs& op_args, string_view key, CmdArgList values,
-                                     bool skip_if_exists) {
+OpResult<uint32_t> OpSet(const OpArgs& op_args, string_view key, CmdArgList values,
+                         bool skip_if_exists) {
   DCHECK(!values.empty() && 0 == values.size() % 2);
 
   auto& db_slice = op_args.shard->db_slice();
   pair<PrimeIterator, bool> add_res;
   try {
     add_res = db_slice.AddOrFind(op_args.db_cntx, key);
-  } catch(bad_alloc&) {
+  } catch (bad_alloc&) {
     return OpStatus::OUT_OF_MEMORY;
   }
 
@@ -514,418 +589,340 @@ OpResult<uint32_t> HSetFamily::OpSet(const OpArgs& op_args, string_view key, Cmd
   return created;
 }
 
-OpResult<uint32_t> HSetFamily::OpDel(const OpArgs& op_args, string_view key, CmdArgList values) {
-  DCHECK(!values.empty());
-
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+}  // namespace
 
-  if (!it_res)
-    return it_res.status();
+void HSetFamily::HDel(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
 
-  db_slice.PreUpdate(op_args.db_cntx.db_index, *it_res);
-  CompactObj& co = (*it_res)->second;
-  robj* hset = co.AsRObj();
-  unsigned deleted = 0;
-  bool key_remove = false;
-  DbTableStats* stats = db_slice.MutableStats(op_args.db_cntx.db_index);
+  args.remove_prefix(2);
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpDel(t->GetOpArgs(shard), key, args);
+  };
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    stats->listpack_bytes -= lpBytes((uint8_t*)hset->ptr);
+  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result || result.status() == OpStatus::KEY_NOTFOUND) {
+    (*cntx)->SendLong(*result);
+  } else {
+    (*cntx)->SendError(result.status());
   }
+}
 
-  for (auto s : values) {
-    op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, s.data(), s.size());
-
-    if (hashTypeDelete(hset, op_args.shard->tmp_str1)) {
-      ++deleted;
-      if (hashTypeLength(hset) == 0) {
-        key_remove = true;
-        break;
-      }
-    }
-  }
+void HSetFamily::HLen(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
 
-  co.SyncRObj();
+  auto cb = [&](Transaction* t, EngineShard* shard) { return OpLen(t->GetOpArgs(shard), key); };
 
-  db_slice.PostUpdate(op_args.db_cntx.db_index, *it_res, key);
-  if (key_remove) {
-    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-      stats->listpack_blob_cnt--;
-    }
-    db_slice.Del(op_args.db_cntx.db_index, *it_res);
-  } else if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    stats->listpack_bytes += lpBytes((uint8_t*)hset->ptr);
+  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    (*cntx)->SendLong(*result);
+  } else {
+    (*cntx)->SendError(result.status());
   }
-
-  return deleted;
 }
 
-auto HSetFamily::OpMGet(const OpArgs& op_args, std::string_view key, CmdArgList fields)
-    -> OpResult<vector<OptStr>> {
-  DCHECK(!fields.empty());
+void HSetFamily::HExists(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  string_view field = ArgS(args, 2);
 
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+  auto cb = [&](Transaction* t, EngineShard* shard) -> OpResult<int> {
+    auto& db_slice = shard->db_slice();
+    auto it_res = db_slice.Find(t->db_context(), key, OBJ_HASH);
 
-  if (!it_res)
+    if (it_res) {
+      robj* hset = (*it_res)->second.AsRObj();
+      shard->tmp_str1 = sdscpylen(shard->tmp_str1, field.data(), field.size());
+
+      return hashTypeExists(hset, shard->tmp_str1);
+    }
+    if (it_res.status() == OpStatus::KEY_NOTFOUND)
+      return 0;
     return it_res.status();
+  };
 
-  CompactObj& co = (*it_res)->second;
-  robj* hset = co.AsRObj();
+  OpResult<int> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    (*cntx)->SendLong(*result);
+  } else {
+    (*cntx)->SendError(result.status());
+  }
+}
 
-  std::vector<OptStr> result(fields.size());
+void HSetFamily::HMGet(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    uint8_t* lp = (uint8_t*)hset->ptr;
-    absl::flat_hash_map<string_view, unsigned> reverse;
-    reverse.reserve(fields.size() + 1);
-    for (size_t i = 0; i < fields.size(); ++i) {
-      reverse.emplace(ArgS(fields, i), i);  // map fields to their index.
-    }
+  args.remove_prefix(2);
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpMGet(t->GetOpArgs(shard), key, args);
+  };
 
-    char ibuf[32];
-    uint8_t* lp_elem = lpFirst(lp);
-    int64_t ele_len;
-    string_view key;
-    DCHECK(lp_elem);  // empty containers are not allowed.
-    size_t lplen = lpLength(lp);
-    DCHECK(lplen > 0 && lplen % 2 == 0);
+  OpResult<vector<OptStr>> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
 
-    // We do single pass on listpack for this operation.
-    do {
-      uint8_t* elem = lpGet(lp_elem, &ele_len, NULL);
-      if (elem) {
-        key = string_view{reinterpret_cast<char*>(elem), size_t(ele_len)};
+  if (result) {
+    (*cntx)->StartArray(result->size());
+    for (const auto& val : *result) {
+      if (val) {
+        (*cntx)->SendBulkString(*val);
       } else {
-        char* next = absl::numbers_internal::FastIntToBuffer(ele_len, ibuf);
-        key = string_view{ibuf, size_t(next - ibuf)};
+        (*cntx)->SendNull();
       }
-      lp_elem = lpNext(lp, lp_elem);  // switch to value
-      DCHECK(lp_elem);
+    }
+  } else if (result.status() == OpStatus::KEY_NOTFOUND) {
+    (*cntx)->StartArray(args.size());
+    for (unsigned i = 0; i < args.size(); ++i) {
+      (*cntx)->SendNull();
+    }
+  } else {
+    (*cntx)->SendError(result.status());
+  }
+}
 
-      auto it = reverse.find(key);
-      if (it != reverse.end()) {
-        DCHECK_LT(it->second, result.size());
-        result[it->second].emplace(LpGetVal(lp_elem));  // populate found items.
-      }
+void HSetFamily::HGet(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  string_view field = ArgS(args, 2);
 
-      lp_elem = lpNext(lp, lp_elem);  // switch to the next key
-    } while (lp_elem);
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpGet(t->GetOpArgs(shard), key, field);
+  };
+
+  OpResult<string> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    (*cntx)->SendBulkString(*result);
   } else {
-    DCHECK_EQ(OBJ_ENCODING_HT, hset->encoding);
-    dict* d = (dict*)hset->ptr;
-    for (size_t i = 0; i < fields.size(); ++i) {
-      op_args.shard->tmp_str1 =
-          sdscpylen(op_args.shard->tmp_str1, fields[i].data(), fields[i].size());
-      dictEntry* de = dictFind(d, op_args.shard->tmp_str1);
-      if (de) {
-        sds val = (sds)dictGetVal(de);
-        result[i].emplace(val, sdslen(val));
-      }
+    if (result.status() == OpStatus::KEY_NOTFOUND) {
+      (*cntx)->SendNull();
+    } else {
+      (*cntx)->SendError(result.status());
     }
   }
-
-  return result;
 }
 
-OpResult<uint32_t> HSetFamily::OpLen(const OpArgs& op_args, string_view key) {
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+void HSetFamily::HIncrBy(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  string_view field = ArgS(args, 2);
+  string_view incrs = ArgS(args, 3);
+  int64_t ival = 0;
 
-  if (it_res) {
-    robj* hset = (*it_res)->second.AsRObj();
-    return hashTypeLength(hset);
+  if (!absl::SimpleAtoi(incrs, &ival)) {
+    return (*cntx)->SendError(kInvalidIntErr);
   }
-  if (it_res.status() == OpStatus::KEY_NOTFOUND)
-    return 0;
-  return it_res.status();
-}
-
-OpResult<string> HSetFamily::OpGet(const OpArgs& op_args, string_view key, string_view field) {
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
-  if (!it_res)
-    return it_res.status();
 
-  robj* hset = (*it_res)->second.AsRObj();
+  IncrByParam param{ival};
 
-  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpIncrBy(t->GetOpArgs(shard), key, field, &param);
+  };
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    unsigned char* vstr = NULL;
-    unsigned int vlen = UINT_MAX;
-    long long vll = LLONG_MAX;
+  OpStatus status = cntx->transaction->ScheduleSingleHop(std::move(cb));
 
-    int ret = hashTypeGetFromListpack(hset, op_args.shard->tmp_str1, &vstr, &vlen, &vll);
-    if (ret < 0) {
-      return OpStatus::KEY_NOTFOUND;
-    }
-    if (vstr) {
-      const char* src = reinterpret_cast<const char*>(vstr);
-      return string{src, vlen};
+  if (status == OpStatus::OK) {
+    (*cntx)->SendLong(get<int64_t>(param));
+  } else {
+    switch (status) {
+      case OpStatus::INVALID_VALUE:
+        (*cntx)->SendError("hash value is not an integer");
+        break;
+      case OpStatus::OUT_OF_RANGE:
+        (*cntx)->SendError(kIncrOverflow);
+        break;
+      default:
+        (*cntx)->SendError(status);
+        break;
     }
-
-    return absl::StrCat(vll);
   }
-  DCHECK_EQ(hset->encoding, OBJ_ENCODING_HT);
-
-  dictEntry* de = dictFind((dict*)hset->ptr, op_args.shard->tmp_str1);
-  if (!de)
-    return OpStatus::KEY_NOTFOUND;
-
-  sds val = (sds)dictGetVal(de);
-
-  return string(val, sdslen(val));
 }
 
-OpResult<vector<string>> HSetFamily::OpGetAll(const OpArgs& op_args, string_view key,
-                                              uint8_t mask) {
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
-  if (!it_res) {
-    if (it_res.status() == OpStatus::KEY_NOTFOUND)
-      return vector<string>{};
-    return it_res.status();
+void HSetFamily::HIncrByFloat(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  string_view field = ArgS(args, 2);
+  string_view incrs = ArgS(args, 3);
+  double dval = 0;
+
+  if (!absl::SimpleAtod(incrs, &dval)) {
+    return (*cntx)->SendError(kInvalidFloatErr);
   }
 
-  robj* hset = (*it_res)->second.AsRObj();
-  hashTypeIterator* hi = hashTypeInitIterator(hset);
+  IncrByParam param{dval};
 
-  vector<string> res;
-  bool keyval = (mask == (FIELDS | VALUES));
-  size_t len = hashTypeLength(hset);
-  res.resize(keyval ? len * 2 : len);
-  unsigned index = 0;
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpIncrBy(t->GetOpArgs(shard), key, field, &param);
+  };
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    while (hashTypeNext(hi) != C_ERR) {
-      if (mask & FIELDS) {
-        res[index++] = LpGetVal(hi->fptr);
-      }
+  OpStatus status = cntx->transaction->ScheduleSingleHop(std::move(cb));
 
-      if (mask & VALUES) {
-        res[index++] = LpGetVal(hi->vptr);
-      }
-    }
+  if (status == OpStatus::OK) {
+    (*cntx)->SendDouble(get<double>(param));
   } else {
-    while (hashTypeNext(hi) != C_ERR) {
-      if (mask & FIELDS) {
-        sds key = (sds)dictGetKey(hi->de);
-        res[index++].assign(key, sdslen(key));
-      }
-
-      if (mask & VALUES) {
-        sds val = (sds)dictGetVal(hi->de);
-        res[index++].assign(val, sdslen(val));
-      }
+    switch (status) {
+      case OpStatus::INVALID_VALUE:
+        (*cntx)->SendError("hash value is not a float");
+        break;
+      default:
+        (*cntx)->SendError(status);
+        break;
     }
   }
+}
 
-  hashTypeReleaseIterator(hi);
-
-  return res;
+void HSetFamily::HKeys(CmdArgList args, ConnectionContext* cntx) {
+  HGetGeneric(args, cntx, FIELDS);
 }
 
-OpResult<size_t> HSetFamily::OpStrLen(const OpArgs& op_args, string_view key, string_view field) {
-  auto& db_slice = op_args.shard->db_slice();
-  auto it_res = db_slice.Find(op_args.db_cntx, key, OBJ_HASH);
+void HSetFamily::HVals(CmdArgList args, ConnectionContext* cntx) {
+  HGetGeneric(args, cntx, VALUES);
+}
 
-  if (!it_res) {
-    if (it_res.status() == OpStatus::KEY_NOTFOUND)
-      return 0;
-    return it_res.status();
-  }
+void HSetFamily::HGetAll(CmdArgList args, ConnectionContext* cntx) {
+  HGetGeneric(args, cntx, GetAllMode::FIELDS | GetAllMode::VALUES);
+}
 
-  robj* hset = (*it_res)->second.AsRObj();
-  size_t field_len = 0;
-  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
+void HSetFamily::HGetGeneric(CmdArgList args, ConnectionContext* cntx, uint8_t getall_mask) {
+  string_view key = ArgS(args, 1);
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    unsigned char* vstr = NULL;
-    unsigned int vlen = UINT_MAX;
-    long long vll = LLONG_MAX;
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpGetAll(t->GetOpArgs(shard), key, getall_mask);
+  };
 
-    if (hashTypeGetFromListpack(hset, op_args.shard->tmp_str1, &vstr, &vlen, &vll) == 0)
-      field_len = vstr ? vlen : sdigits10(vll);
+  OpResult<vector<string>> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
 
-    return field_len;
+  if (result) {
+    (*cntx)->SendStringArr(absl::Span<const string>{*result});
+  } else {
+    (*cntx)->SendError(result.status());
   }
-
-  DCHECK_EQ(hset->encoding, OBJ_ENCODING_HT);
-
-  dictEntry* de = dictFind((dict*)hset->ptr, op_args.shard->tmp_str1);
-  return de ? sdslen((sds)de->v.val) : 0;
 }
 
-OpStatus HSetFamily::OpIncrBy(const OpArgs& op_args, string_view key, string_view field,
-                              IncrByParam* param) {
-  auto& db_slice = op_args.shard->db_slice();
-  const auto [it, inserted] = db_slice.AddOrFind(op_args.db_cntx, key);
-
-  DbTableStats* stats = db_slice.MutableStats(op_args.db_cntx.db_index);
+void HSetFamily::HScan(CmdArgList args, ConnectionContext* cntx) {
+  std::string_view key = ArgS(args, 1);
+  std::string_view token = ArgS(args, 2);
 
-  robj* hset = nullptr;
-  size_t lpb = 0;
+  uint64_t cursor = 0;
 
-  if (inserted) {
-    hset = createHashObject();
-    it->second.ImportRObj(hset);
-    stats->listpack_blob_cnt++;
-    hset = it->second.AsRObj();
-  } else {
-    if (it->second.ObjType() != OBJ_HASH)
-      return OpStatus::WRONG_TYPE;
+  if (!absl::SimpleAtoi(token, &cursor)) {
+    return (*cntx)->SendError("invalid cursor");
+  }
 
-    db_slice.PreUpdate(op_args.db_cntx.db_index, it);
-    hset = it->second.AsRObj();
+  if (args.size() > 3) {
+    return (*cntx)->SendError("scan options are not supported yet");
+  }
 
-    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-      lpb = lpBytes((uint8_t*)hset->ptr);
-      stats->listpack_bytes -= lpb;
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpScan(t->GetOpArgs(shard), key, &cursor);
+  };
 
-      if (lpb >= kMaxListPackLen) {
-        stats->listpack_blob_cnt--;
-        hashTypeConvert(hset, OBJ_ENCODING_HT);
-      }
+  OpResult<StringVec> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result.status() != OpStatus::WRONG_TYPE) {
+    (*cntx)->StartArray(2);
+    (*cntx)->SendSimpleString(absl::StrCat(cursor));
+    (*cntx)->StartArray(result->size());
+    for (const auto& k : *result) {
+      (*cntx)->SendBulkString(k);
     }
+  } else {
+    (*cntx)->SendError(result.status());
   }
+}
 
-  unsigned char* vstr = NULL;
-  unsigned int vlen = UINT_MAX;
-  long long old_val = 0;
-
-  op_args.shard->tmp_str1 = sdscpylen(op_args.shard->tmp_str1, field.data(), field.size());
-
-  int exist_res = hashTypeGetValue(hset, op_args.shard->tmp_str1, &vstr, &vlen, &old_val);
+void HSetFamily::HSet(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  ToLower(&args[0]);
 
-  if (holds_alternative<double>(*param)) {
-    long double value;
-    double incr = get<double>(*param);
-    if (exist_res == C_OK) {
-      if (vstr) {
-        const char* exist_val = reinterpret_cast<char*>(vstr);
+  string_view cmd = ArgS(args, 0);
 
-        if (!string2ld(exist_val, vlen, &value)) {
-          stats->listpack_bytes += lpb;
+  if (args.size() % 2 != 0) {
+    return (*cntx)->SendError(facade::WrongNumArgsError(cmd), kSyntaxErrType);
+  }
 
-          return OpStatus::INVALID_VALUE;
-        }
-      } else {
-        value = old_val;
-      }
-      value += incr;
+  args.remove_prefix(2);
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpSet(t->GetOpArgs(shard), key, args, false);
+  };
 
-      if (isnan(value) || isinf(value)) {
-        return OpStatus::INVALID_FLOAT;
-      }
-    } else {
-      value = incr;
-    }
+  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result && cmd == "hset") {
+    (*cntx)->SendLong(*result);
+  } else {
+    (*cntx)->SendError(result.status());
+  }
+}
 
-    char buf[128];
-    char* str = RedisReplyBuilder::FormatDouble(value, buf, sizeof(buf));
-    string_view sval{str};
+void HSetFamily::HSetNx(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
 
-    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-      uint8_t* lp = (uint8_t*)hset->ptr;
+  args.remove_prefix(2);
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpSet(t->GetOpArgs(shard), key, args, true);
+  };
 
-      lp = LpInsert(lp, field, sval, false).first;
-      hset->ptr = lp;
-      stats->listpack_bytes += lpBytes(lp);
-    } else {
-      sds news = sdsnewlen(str, sval.size());
-      hashTypeSet(hset, op_args.shard->tmp_str1, news, HASH_SET_TAKE_VALUE);
-    }
-    param->emplace<double>(value);
+  OpResult<uint32_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    (*cntx)->SendLong(*result);
   } else {
-    if (exist_res == C_OK && vstr) {
-      const char* exist_val = reinterpret_cast<char*>(vstr);
-      if (!string2ll(exist_val, vlen, &old_val)) {
-        stats->listpack_bytes += lpb;
-
-        return OpStatus::INVALID_VALUE;
-      }
-    }
-    int64_t incr = get<int64_t>(*param);
-    if ((incr < 0 && old_val < 0 && incr < (LLONG_MIN - old_val)) ||
-        (incr > 0 && old_val > 0 && incr > (LLONG_MAX - old_val))) {
-      stats->listpack_bytes += lpb;
-
-      return OpStatus::OUT_OF_RANGE;
-    }
+    (*cntx)->SendError(result.status());
+  }
+}
 
-    int64_t new_val = old_val + incr;
+void HSetFamily::HStrLen(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
+  string_view field = ArgS(args, 2);
 
-    if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-      char buf[32];
-      char* next = absl::numbers_internal::FastIntToBuffer(new_val, buf);
-      string_view sval{buf, size_t(next - buf)};
-      uint8_t* lp = (uint8_t*)hset->ptr;
+  auto cb = [&](Transaction* t, EngineShard* shard) {
+    return OpStrLen(t->GetOpArgs(shard), key, field);
+  };
 
-      lp = LpInsert(lp, field, sval, false).first;
-      hset->ptr = lp;
-      stats->listpack_bytes += lpBytes(lp);
-    } else {
-      sds news = sdsfromlonglong(new_val);
-      hashTypeSet(hset, op_args.shard->tmp_str1, news, HASH_SET_TAKE_VALUE);
-    }
-    param->emplace<int64_t>(new_val);
+  OpResult<size_t> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    (*cntx)->SendLong(*result);
+  } else {
+    (*cntx)->SendError(result.status());
   }
-
-  it->second.SyncRObj();
-  db_slice.PostUpdate(op_args.db_cntx.db_index, it, key);
-
-  return OpStatus::OK;
 }
 
-OpResult<StringVec> HSetFamily::OpScan(const OpArgs& op_args, std::string_view key,
-                                       uint64_t* cursor) {
-  OpResult<PrimeIterator> find_res = op_args.shard->db_slice().Find(op_args.db_cntx, key, OBJ_HASH);
+void HSetFamily::HRandField(CmdArgList args, ConnectionContext* cntx) {
+  string_view key = ArgS(args, 1);
 
-  if (!find_res)
-    return find_res.status();
+  auto cb = [&](Transaction* t, EngineShard* shard) -> OpResult<StringVec> {
+    auto& db_slice = shard->db_slice();
+    auto it_res = db_slice.Find(t->db_context(), key, OBJ_HASH);
 
-  PrimeIterator it = find_res.value();
-  StringVec res;
-  uint32_t count = 20;
-  robj* hset = it->second.AsRObj();
+    if (!it_res)
+      return it_res.status();
 
-  if (hset->encoding == OBJ_ENCODING_LISTPACK) {
-    uint8_t* lp = (uint8_t*)hset->ptr;
-    uint8_t* lp_elem = lpFirst(lp);
+    const PrimeValue& pv = it_res.value()->second;
+    StringVec str_vec;
 
-    DCHECK(lp_elem);  // empty containers are not allowed.
+    if (pv.Encoding() == OBJ_ENCODING_HT) {
+      dict* this_dict = (dict*)pv.RObjPtr();
+      dictEntry* de = dictGetFairRandomKey(this_dict);
+      sds key = (sds)de->key;
+      str_vec.emplace_back(key, sdslen(key));
+    } else if (pv.Encoding() == OBJ_ENCODING_LISTPACK) {
+      uint8_t* lp = (uint8_t*)pv.RObjPtr();
+      size_t lplen = lpLength(lp);
+      CHECK(lplen > 0 && lplen % 2 == 0);
 
-    int64_t ele_len;
-    unsigned char intbuf[LP_INTBUF_SIZE];
+      size_t hlen = lplen / 2;
+      listpackEntry key;
 
-    // We do single pass on listpack for this operation.
-    do {
-      uint8_t* elem = lpGet(lp_elem, &ele_len, intbuf);
-      DCHECK(elem);
-      res.emplace_back(reinterpret_cast<char*>(elem), size_t(ele_len));
-      lp_elem = lpNext(lp, lp_elem);  // switch to value
-    } while (lp_elem);
-    *cursor = 0;
-  } else {
-    dict* ht = (dict*)hset->ptr;
-    long maxiterations = count * 10;
-    void* privdata = &res;
-    auto scanCb = [](void* privdata, const dictEntry* de) {
-      StringVec* res = (StringVec*)privdata;
-      sds val = (sds)de->key;
-      res->emplace_back(val, sdslen(val));
-      val = (sds)de->v.val;
-      res->emplace_back(val, sdslen(val));
-    };
+      lpRandomPair(lp, hlen, &key, NULL);
+      if (key.sval) {
+        str_vec.emplace_back(reinterpret_cast<char*>(key.sval), key.slen);
+      } else {
+        str_vec.emplace_back(absl::StrCat(key.lval));
+      }
+    } else {
+      LOG(ERROR) << "Invalid encoding " << pv.Encoding();
+      return OpStatus::INVALID_VALUE;
+    }
+    return str_vec;
+  };
 
-    do {
-      *cursor = dictScan(ht, *cursor, scanCb, NULL, privdata);
-    } while (*cursor && maxiterations-- && res.size() < count);
+  OpResult<StringVec> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
+  if (result) {
+    CHECK_EQ(1u, result->size());  // TBD: to support count and withvalues.
+    (*cntx)->SendBulkString(result->front());
+  } else {
+    (*cntx)->SendError(result.status());
   }
-
-  return res;
 }
 
 using CI = CommandId;
diff --git a/src/server/hset_family.h b/src/server/hset_family.h
index d45d378cf675..e9fbcf2a05a1 100644
--- a/src/server/hset_family.h
+++ b/src/server/hset_family.h
@@ -22,7 +22,6 @@ class HSetFamily {
   static uint32_t MaxListPackLen();
 
  private:
-  enum GetAllMode : uint8_t { FIELDS = 1, VALUES = 2 };
 
   static void HDel(CmdArgList args, ConnectionContext* cntx);
   static void HLen(CmdArgList args, ConnectionContext* cntx);
@@ -41,30 +40,6 @@ class HSetFamily {
   static void HRandField(CmdArgList args, ConnectionContext* cntx);
 
   static void HGetGeneric(CmdArgList args, ConnectionContext* cntx, uint8_t getall_mask);
-
-  static OpResult<uint32_t> OpSet(const OpArgs& op_args, std::string_view key, CmdArgList values,
-                                  bool skip_if_exists);
-  static OpResult<uint32_t> OpDel(const OpArgs& op_args, std::string_view key, CmdArgList values);
-
-  using OptStr = std::optional<std::string>;
-  static OpResult<std::vector<OptStr>> OpMGet(const OpArgs& op_args, std::string_view key,
-                                              CmdArgList fields);
-
-  static OpResult<uint32_t> OpLen(const OpArgs& op_args, std::string_view key);
-
-  static OpResult<std::string> OpGet(const OpArgs& op_args, std::string_view key,
-                                     std::string_view field);
-
-  static OpResult<std::vector<std::string>> OpGetAll(const OpArgs& op_args, std::string_view key,
-                                                     uint8_t getall_mask);
-  static OpResult<size_t> OpStrLen(const OpArgs& op_args, std::string_view key,
-                                   std::string_view field);
-
-  using IncrByParam = std::variant<double, int64_t>;
-  static OpStatus OpIncrBy(const OpArgs& op_args, std::string_view key, std::string_view field,
-                           IncrByParam* param);
-
-  static OpResult<StringVec> OpScan(const OpArgs& op_args, std::string_view key, uint64_t* cursor);
 };
 
 }  // namespace dfly