Convert quicklist RDB to store ziplist nodes

Turns out it's a huge improvement during save/reload/migrate/restore
because, with compression enabled, we're compressing 4k or 8k
chunks of data consisting of multiple elements in one ziplist
instead of compressing series of smaller individual elements.

Author: mattsta

src/rdb.c

@@ -434,7 +434,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
         return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
     case REDIS_LIST:
         if (o->encoding == REDIS_ENCODING_QUICKLIST)
-            return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST);
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_QUICKLIST);
         else
             redisPanic("Unknown list encoding");
     case REDIS_SET:
@@ -484,22 +484,16 @@ int rdbSaveObject(rio *rdb, robj *o) {
     } else if (o->type == REDIS_LIST) {
         /* Save a list value */
         if (o->encoding == REDIS_ENCODING_QUICKLIST) {
-            quicklist *list = o->ptr;
-            quicklistIter *li = quicklistGetIterator(list, AL_START_HEAD);
-            quicklistEntry entry;
+            quicklist *ql = o->ptr;
+            quicklistNode *node = ql->head;
 
-            if ((n = rdbSaveLen(rdb,quicklistCount(list))) == -1) return -1;
+            if ((n = rdbSaveLen(rdb,ql->len)) == -1) return -1;
             nwritten += n;
 
-            while (quicklistNext(li,&entry)) {
-                if (entry.value) {
-                    if ((n = rdbSaveRawString(rdb,entry.value,entry.sz)) == -1) return -1;
-                } else {
-                    if ((n = rdbSaveLongLongAsStringObject(rdb,entry.longval)) == -1) return -1;
-                }
+            do {
+                if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1;
                 nwritten += n;
-            }
-            quicklistReleaseIterator(li);
+            } while ((node = node->next));
         } else {
             redisPanic("Unknown list encoding");
         }
@@ -974,7 +968,19 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
 
         /* All pairs should be read by now */
         redisAssert(len == 0);
+    } else if (rdbtype == REDIS_RDB_TYPE_LIST_QUICKLIST) {
+        if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
+        o = createQuicklistObject();
 
+        while (len--) {
+            if ((ele = rdbLoadStringObject(rdb)) == NULL) return NULL;
+            /* 'ele' contains a sds of the ziplist, but we need to extract
+             * the actual ziplist for future usage. We must copy the
+             * sds contents to a new buffer. */
+            unsigned char *zl = (unsigned char *)sdsnative(ele->ptr);
+            zfree(ele); /* free robj container since we keep the ziplist */
+            quicklistAppendZiplist(o->ptr, zl);
+        }
     } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP  ||
                rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
                rdbtype == REDIS_RDB_TYPE_SET_INTSET   ||

src/rdb.h

@@ -38,7 +38,7 @@
 
 /* The current RDB version. When the format changes in a way that is no longer
  * backward compatible this number gets incremented. */
-#define REDIS_RDB_VERSION 6
+#define REDIS_RDB_VERSION 7
 
 /* Defines related to the dump file format. To store 32 bits lengths for short
  * keys requires a lot of space, so we check the most significant 2 bits of
@@ -74,16 +74,19 @@
 #define REDIS_RDB_TYPE_SET    2
 #define REDIS_RDB_TYPE_ZSET   3
 #define REDIS_RDB_TYPE_HASH   4
+/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
 
 /* Object types for encoded objects. */
 #define REDIS_RDB_TYPE_HASH_ZIPMAP    9
 #define REDIS_RDB_TYPE_LIST_ZIPLIST  10
 #define REDIS_RDB_TYPE_SET_INTSET    11
 #define REDIS_RDB_TYPE_ZSET_ZIPLIST  12
 #define REDIS_RDB_TYPE_HASH_ZIPLIST  13
+#define REDIS_RDB_TYPE_LIST_QUICKLIST 14
+/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
 
 /* Test if a type is an object type. */
-#define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 13))
+#define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 14))
 
 /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
 #define REDIS_RDB_OPCODE_EXPIRETIME_MS 252