feat(rdb): add DUMP support for SortedInt type (#3366)

- Serialize SortedInt as RDB Set for Redis compatibility
- Use RangeByValue() to get all integers deterministically
- No RESTORE changes (SortedInt dumps restore as Set, like
Bitmap→String)
- Resolves part of #3319

Testing:
- Manual test confirms DUMP no longer returns error
- RESTORE successfully creates Set with preserved data
- All integers maintained in restored Set

Manuel test:
root@codespaces-6e5713:/workspaces/kvrocks# # Connect to Kvrocks
redis-cli -p 6666

127.0.0.1:6666> SIADD testkey 100 200 300
(integer) 0 // already exists

127.0.0.1:6666> SICARD testkey
(integer) 3

127.0.0.1:6666> TYPE testkey
sortedint

127.0.0.1:6666> DUMP testkey

"\x02\x03\xc0d\xc1\xc8\x00\xc1,\x01\x06\x00\xb2\x01\xaa\xe2\x06h\xba\x8d"

127.0.0.1:6666> SIADD mykey 5 12 23 89 100
(integer) 0
127.0.0.1:6666> DUMP mykey

"\x02\x05\xc0\x05\xc0\x0c\xc0\x17\xc0Y\xc0d\x06\x00\xa8\x19U\x90\x8f\x011o"

127.0.0.1:6666> RESTORE newkey 0
"\x02\x05\xc0\x05\xc0\x0c\xc0\x17\xc0Y\xc0d\x06\x00\xa8\x19U\x90\x8f\x011o"
OK

127.0.0.1:6666> TYPE newkey
set

127.0.0.1:6666> SMEMBERS newkey
1) "100"
2) "12"
3) "23"
4) "5"
5) "89"
127.0.0.1:6666> DUMP nonexistent
(nil)
127.0.0.1:6666> exit

---------

Co-authored-by: hulk <hulk.website@gmail.com>

Author: Zakir032002

src/storage/rdb/rdb.cc

@@ -20,6 +20,8 @@
 
 #include "rdb.h"
 
+#include <limits>
+
 #include "common/encoding.h"
 #include "common/rdb_stream.h"
 #include "common/time_util.h"
@@ -35,6 +37,7 @@
 #include "types/redis_hash.h"
 #include "types/redis_list.h"
 #include "types/redis_set.h"
+#include "types/redis_sortedint.h"
 #include "types/redis_string.h"
 #include "types/redis_zset.h"
 #include "vendor/crc64.h"
@@ -730,6 +733,8 @@ Status RDB::SaveObjectType(const RedisType type) {
     robj_type = RDBTypeSet;
   } else if (type == kRedisZSet) {
     robj_type = RDBTypeZSet2;
+  } else if (type == kRedisSortedint) {
+    robj_type = RDBTypeSet;
   } else {
     WARN("Invalid or Not supported object type: {}", (int)type);
     return {Status::NotOK, "Invalid or Not supported object type"};
@@ -793,6 +798,23 @@ Status RDB::SaveObject(const std::string &key, const RedisType type) {
       return {Status::RedisExecErr, s.ToString()};
     }
     return SaveStringObject(value);
+  } else if (type == kRedisSortedint) {
+    redis::Sortedint sortedint_db(storage_, ns_);
+    std::vector<uint64_t> ids;
+    SortedintRangeSpec spec;
+    spec.min = 0;
+    spec.max = std::numeric_limits<uint64_t>::max();
+    spec.minex = false;
+    spec.maxex = false;
+    spec.offset = 0;
+    spec.count = std::numeric_limits<int>::max();
+    spec.reversed = false;
+    int size = 0;
+    auto si_status = sortedint_db.RangeByValue(ctx, key, spec, &ids, &size);
+    if (!si_status.ok()) {
+      return {Status::RedisExecErr, si_status.ToString()};
+    }
+    return SaveSortedintObject(ids);
   } else {
     WARN("Invalid or Not supported object type: {}", (int)type);
     return {Status::NotOK, "Invalid or Not supported object type"};
@@ -930,6 +952,20 @@ Status RDB::SaveHashObject(const std::vector<FieldValue> &field_values) {
   }
   return Status::OK();
 }
+Status RDB::SaveSortedintObject(const std::vector<uint64_t> &ids) {
+  if (ids.size() > 0) {
+    auto status = RdbSaveLen(ids.size());
+    if (!status.IsOK()) return status;
+
+    for (const auto &id : ids) {
+      status = SaveStringObject(std::to_string(id));
+      if (!status.IsOK()) return status;
+    }
+  } else {
+    return {Status::NotOK, "the size of sortedint is zero"};
+  }
+  return Status::OK();
+}
 
 int RDB::rdbEncodeInteger(const long long value, unsigned char *enc) {
   if (value >= -(1 << 7) && value <= (1 << 7) - 1) {

src/storage/rdb/rdb.h

@@ -130,6 +130,9 @@ class RDB {
   // Hash
   Status SaveHashObject(const std::vector<FieldValue> &field_value);
 
+  // SortedInt
+  Status SaveSortedintObject(const std::vector<uint64_t> &ids);
+
  private:
   engine::Storage *storage_;
   std::string ns_;

tests/gocase/unit/dump/dump_test.go

@@ -193,3 +193,30 @@ func TestDump_IntegerEncoding(t *testing.T) {
 		require.Equal(t, v, got)
 	}
 }
+
+func TestDump_SortedInt(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	key := "test_sortedint_key"
+	require.NoError(t, rdb.Del(ctx, key).Err())
+
+	require.EqualValues(t, 5, rdb.Do(ctx, "SIADD", key, 5, 12, 23, 89, 100).Val())
+	require.EqualValues(t, 5, rdb.Do(ctx, "SICARD", key).Val())
+	require.EqualValues(t, "sortedint", rdb.Type(ctx, key).Val())
+
+	serialized, err := rdb.Dump(ctx, key).Result()
+	require.NoError(t, err)
+
+	restoredKey := fmt.Sprintf("restore_%s", key)
+	require.NoError(t, rdb.RestoreReplace(ctx, restoredKey, 0, serialized).Err())
+
+	require.EqualValues(t, "set", rdb.Type(ctx, restoredKey).Val())
+	members := rdb.SMembers(ctx, restoredKey).Val()
+	expectedMembers := []string{"5", "12", "23", "89", "100"}
+	require.ElementsMatch(t, expectedMembers, members)
+}