PragmaTwice commented on code in PR #2227:
URL: https://github.com/apache/kvrocks/pull/2227#discussion_r1554831158
##########
src/storage/rdb.cc:
##########
@@ -680,3 +679,323 @@ Status RDB::LoadRdb(uint32_t db_index, bool
overwrite_exist_key) {
return Status::OK();
}
+
+Status RDB::Dump(const std::string &key, const RedisType type) {
+ unsigned char buf[2];
+ /* Serialize the object in an RDB-like format. It consist of an object type
+ * byte followed by the serialized object. This is understood by RESTORE. */
+ auto s = SaveObjectType(type);
+ if (!s.IsOK()) return {Status::RedisExecErr, s.Msg()};
+ s = SaveObject(key, type);
+ if (!s.IsOK()) return {Status::RedisExecErr, s.Msg()};
+
+ /* Write the footer, this is how it looks like:
+ * ----------------+---------------------+---------------+
+ * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 |
+ * ----------------+---------------------+---------------+
+ * RDB version and CRC are both in little endian.
+ */
+
+ /* RDB version */
+ buf[0] = MaxRDBVersion & 0xff;
+ buf[1] = (MaxRDBVersion >> 8) & 0xff;
+ s = stream_->Write((const char *)buf, 2);
+ if (!s.IsOK()) {
+ return {Status::RedisExecErr, s.Msg()};
+ }
+
+ /* CRC64 */
+ std::string &output = static_cast<RdbStringStream
*>(stream_.get())->GetInput();
+ uint64_t crc = crc64(0, (unsigned char *)(output.c_str()), output.length());
+ memrev64ifbe(&crc);
+ s = stream_->Write((const char *)(&crc), 8);
+ if (!s.IsOK()) {
+ return {Status::RedisExecErr, s.Msg()};
+ }
+
+ return Status::OK();
+}
+
+Status RDB::SaveObjectType(const RedisType type) {
+ int robj_type = -1;
+ if (type == kRedisString) {
+ robj_type = RDBTypeString;
+ } else if (type == kRedisHash) {
+ robj_type = RDBTypeHash;
+ } else if (type == kRedisList) {
+ robj_type = RDBTypeListQuickList2;
+ } else if (type == kRedisSet) {
+ robj_type = RDBTypeSet;
+ } else if (type == kRedisZSet) {
+ robj_type = RDBTypeZSet2;
+ } else {
+ LOG(WARNING) << "Invalid or Not supported object type: " << type;
+ return {Status::NotOK, "Invalid or Not supported object type"};
+ }
+ return stream_->Write((const char *)(&robj_type), 1);
+}
+
+Status RDB::SaveObject(const std::string &key, const RedisType type) {
+ if (type == kRedisString) {
+ std::string value;
+ redis::String string_db(storage_, ns_);
+ auto s = string_db.Get(key, &value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ return SaveStringObject(value);
+ } else if (type == kRedisList) {
+ std::vector<std::string> elems;
+ redis::List list_db(storage_, ns_);
+ auto s = list_db.Range(key, 0, -1, &elems);
+ if (!s.ok() && !s.IsNotFound()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ return SaveListObject(elems);
+ } else if (type == kRedisSet) {
+ redis::Set set_db(storage_, ns_);
+ std::vector<std::string> members;
+ auto s = set_db.Members(key, &members);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ return SaveSetObject(members);
+ } else if (type == kRedisZSet) {
+ redis::ZSet zset_db(storage_, ns_);
+ std::vector<MemberScore> member_scores;
+ RangeScoreSpec spec;
+ auto s = zset_db.RangeByScore(key, spec, &member_scores, nullptr);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ std::sort(member_scores.begin(), member_scores.end(),
+ [](const MemberScore &v1, const MemberScore &v2) { return
v1.score > v2.score; });
+ return SaveZSetObject(member_scores);
+ } else if (type == kRedisHash) {
+ redis::Hash hash_db(storage_, ns_);
+ std::vector<FieldValue> field_values;
+ auto s = hash_db.GetAll(key, &field_values);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+
+ return SaveHashObject(field_values);
+ } else {
+ LOG(WARNING) << "Invalid or Not supported object type: " << type;
+ return {Status::NotOK, "Invalid or Not supported object type"};
+ }
+}
+
+Status RDB::RdbSaveLen(uint64_t len) {
+ unsigned char buf[2];
+ if (len < (1 << 6)) {
+ /* Save a 6 bit len */
+ buf[0] = (len & 0xFF) | (RDB6BitLen << 6);
+ auto status = stream_->Write((const char *)buf, 1);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+ return Status::OK();
+ } else if (len < (1 << 14)) {
+ /* Save a 14 bit len */
+ buf[0] = ((len >> 8) & 0xFF) | (RDB14BitLen << 6);
+ buf[1] = len & 0xFF;
+ auto status = stream_->Write((const char *)buf, 2);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+ return Status::OK();
+ } else if (len <= UINT32_MAX) {
+ /* Save a 32 bit len */
+ buf[0] = RDB32BitLen;
+ auto status = stream_->Write((const char *)buf, 1);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+
+ uint32_t len32 = htonl(len);
+ status = stream_->Write((const char *)(&len32), 4);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+ return Status::OK();
+ } else {
+ /* Save a 64 bit len */
+ buf[0] = RDB64BitLen;
+ auto status = stream_->Write((const char *)buf, 1);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+
+ len = htonu64(len);
+ status = stream_->Write((const char *)(&len), 8);
+ if (!status.IsOK()) {
+ return {Status::RedisExecErr, status.Msg()};
+ }
+ return Status::OK();
+ }
+}
+
+Status RDB::SaveStringObject(const std::string &value) {
+ const size_t len = value.length();
+ int enclen = 0;
+
+ // When the length is less than 11, value may be an integer,
+ // so special encoding is performed.
+ if (len <= 11) {
+ unsigned char buf[5];
+ // convert string to long long
+ std::istringstream iss(value);
+ long long integer_value = 0;
+ if (iss.eof() && !iss.fail()) {
+ iss >> integer_value;
Review Comment:
Could you please use `util::ParseInt` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
