ZSET B+ Tree PR 2: ordered index interface and full conversion

cmake/Modules/SourceFiles.cmake

@@ -34,6 +34,8 @@ set(VALKEY_SERVER_SRCS
     ${CMAKE_SOURCE_DIR}/src/t_set.c
     ${CMAKE_SOURCE_DIR}/src/t_zset.c
     ${CMAKE_SOURCE_DIR}/src/skiplist.c
+    ${CMAKE_SOURCE_DIR}/src/skiplist_ordered_index.c
+    ${CMAKE_SOURCE_DIR}/src/ordered_index.c
     ${CMAKE_SOURCE_DIR}/src/t_hash.c
     ${CMAKE_SOURCE_DIR}/src/config.c
     ${CMAKE_SOURCE_DIR}/src/aof.c

src/Makefile

@@ -555,6 +555,8 @@ ENGINE_SERVER_OBJ = \
     sha1.o \
     sha256.o \
     siphash.o \
+    skiplist_ordered_index.o \
+    ordered_index.o \
     socket.o \
     sort.o \
     sparkline.o \

src/aof.c

@@ -28,7 +28,7 @@
  */
 
 #include "server.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "bio.h"
 #include "rio.h"
 #include "functions.h"
@@ -2027,7 +2027,7 @@ int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
         hashtableInitIterator(&iter, zs->ht, 0);
         void *next;
         while (hashtableNext(&iter, &next)) {
-            zskiplistNode *node = next;
+            OrderedIndexItem *node = next;
             if (count == 0) {
                 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? AOF_REWRITE_ITEMS_PER_CMD : items;
 
@@ -2037,8 +2037,10 @@ int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
                     return 0;
                 }
             }
-            sds ele = zslGetNodeElement(node);
-            if (!rioWriteBulkDouble(r, node->score) || !rioWriteBulkString(r, ele, sdslen(ele))) {
+            const char *ele;
+            size_t ele_len;
+            orderedIndexGetElementRaw(node, &ele, &ele_len);
+            if (!rioWriteBulkDouble(r, orderedIndexGetScore(node)) || !rioWriteBulkString(r, ele, ele_len)) {
                 hashtableCleanupIterator(&iter);
                 return 0;
             }

src/db.c

@@ -28,7 +28,7 @@
  */
 
 #include "server.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "cluster.h"
 #include "cluster_migrateslots.h"
 #include "latency.h"
@@ -1052,6 +1052,8 @@ void hashtableScanCallback(void *privdata, void *entry) {
     scanData *data = (scanData *)privdata;
     stringRef val = {NULL, 0};
     sds key = NULL;
+    const char *zset_ptr = NULL;
+    size_t zset_ele_len = 0;
 
     robj *o = data->o;
     data->sampled++;
@@ -1064,9 +1066,8 @@ void hashtableScanCallback(void *privdata, void *entry) {
     if (o->type == OBJ_SET) {
         key = (sds)entry;
     } else if (o->type == OBJ_ZSET) {
-        zskiplistNode *node = (zskiplistNode *)entry;
-        key = zslGetNodeElement(node);
-        /* zset data is copied after filtering by key */
+        orderedIndexGetElementRaw((const OrderedIndexItem *)entry, &zset_ptr, &zset_ele_len);
+        /* zset data is copied after filtering */
     } else if (o->type == OBJ_HASH) {
         key = entryGetField(entry);
         if (!data->only_keys) {
@@ -1078,7 +1079,9 @@ void hashtableScanCallback(void *privdata, void *entry) {
 
     /* Filter element if it does not match the pattern. */
     if (data->pattern) {
-        if (!stringmatchlen(data->pattern, sdslen(data->pattern), key, sdslen(key), 0)) {
+        const char *match_ptr = (o->type == OBJ_ZSET) ? zset_ptr : key;
+        size_t match_len = (o->type == OBJ_ZSET) ? zset_ele_len : sdslen(key);
+        if (!stringmatchlen(data->pattern, sdslen(data->pattern), match_ptr, match_len, 0)) {
             return;
         }
     }
@@ -1087,11 +1090,13 @@ void hashtableScanCallback(void *privdata, void *entry) {
      * allocations. */
     if (o->type == OBJ_ZSET) {
         /* zset data is copied */
-        zskiplistNode *node = (zskiplistNode *)entry;
-        key = sdsdup(zslGetNodeElement(node));
+        const char *ptr;
+        size_t ele_len;
+        orderedIndexGetElementRaw((const OrderedIndexItem *)entry, &ptr, &ele_len);
+        key = sdsnewlen(ptr, ele_len);
         if (!data->only_keys) {
             char buf[MAX_LONG_DOUBLE_CHARS];
-            int len = ld2string(buf, sizeof(buf), node->score, LD_STR_AUTO);
+            int len = ld2string(buf, sizeof(buf), orderedIndexGetScore((const OrderedIndexItem *)entry), LD_STR_AUTO);
             sds tmp = sdsnewlen(buf, len);
             val.buf = (const char *)tmp;
             val.len = sdslen(tmp);

src/debug.c

@@ -28,7 +28,7 @@
  */
 
 #include "server.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "util.h"
 #include "sha1.h" /* SHA1 is used for DEBUG DIGEST */
 #include "crc64.h"
@@ -217,12 +217,14 @@ void xorObjectDigest(serverDb *db, robj *keyobj, unsigned char *digest, robj *o)
 
             void *next;
             while (hashtableNext(&iter, &next)) {
-                zskiplistNode *node = next;
-                const int len = fpconv_dtoa(node->score, buf);
+                OrderedIndexItem *node = next;
+                const char *ele;
+                size_t ele_len;
+                orderedIndexGetElementRaw(node, &ele, &ele_len);
+                const int len = fpconv_dtoa(orderedIndexGetScore(node), buf);
                 buf[len] = '\0';
                 memset(eledigest, 0, 20);
-                sds ele = zslGetNodeElement(node);
-                mixDigest(eledigest, ele, sdslen(ele));
+                mixDigest(eledigest, ele, ele_len);
                 mixDigest(eledigest, buf, strlen(buf));
                 xorDigest(digest, eledigest, 20);
             }
@@ -1184,8 +1186,11 @@ void serverLogObjectDebugInfo(const robj *o) {
         serverLog(LL_WARNING, "Hash size: %d", (int)hashTypeLength(o));
     } else if (o->type == OBJ_ZSET) {
         serverLog(LL_WARNING, "Sorted set size: %d", (int)zsetLength(o));
-        if (o->encoding == OBJ_ENCODING_SKIPLIST)
-            serverLog(LL_WARNING, "Skiplist level: %d", (int)((const zset *)o->ptr)->zsl->level);
+        if (o->encoding == OBJ_ENCODING_SKIPLIST) {
+            /* Not declared in ordered_index.h — debug-only introspection. */
+            extern int orderedIndexGetDepth(OrderedIndex * oi);
+            serverLog(LL_WARNING, "Index depth: %d", orderedIndexGetDepth(((const zset *)o->ptr)->oi));
+        }
     } else if (o->type == OBJ_STREAM) {
         serverLog(LL_WARNING, "Stream size: %d", (int)streamLength(o));
     }

src/defrag.c

@@ -39,7 +39,7 @@
  */
 
 #include "server.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "hashtable.h"
 #include "eval.h"
 #include "script.h"
@@ -237,61 +237,13 @@ robj *activeDefragStringOb(robj *ob) {
     return new_robj;
 }
 
-/* Internal function used by zslDefrag */
-static void zslUpdateNode(zskiplist *zsl, zskiplistNode *oldnode, zskiplistNode *newnode, zskiplistNode **update) {
-    int i;
-    for (i = 0; i < zslGetHeight(zsl); i++) {
-        if (update[i]->level[i].forward == oldnode) update[i]->level[i].forward = newnode;
-    }
-    serverAssert(zslGetHeader(zsl) != oldnode);
-    if (newnode->level[0].forward) {
-        serverAssert(newnode->level[0].forward->backward == oldnode);
-        newnode->level[0].forward->backward = newnode;
-    } else {
-        serverAssert(zslGetTail(zsl) == oldnode);
-        zslSetTail(zsl, newnode);
-    }
-}
-
-/* Hashtable scan callback for sorted set. It defragments a single skiplist
- * node, updates skiplist pointers, and updates the hashtable pointer to the
- * node. */
-static void activeDefragZsetNode(void *privdata, void *entry_ref) {
-    zskiplist *zsl = privdata;
-    zskiplistNode **node_ref = (zskiplistNode **)entry_ref;
-    zskiplistNode *node = *node_ref;
-
-    size_t allocation_size;
-    zskiplistNode *newnode = activeDefragAllocWithoutFree(node, &allocation_size);
-    if (newnode == NULL) return;
-
-    const double score = node->score;
-
-    /* find skiplist pointers that need to be updated if we end up moving the
-     * skiplist node. */
-    sds ele = zslGetNodeElement(node);
-    zskiplistNode *update[ZSKIPLIST_MAXLEVEL];
-    zskiplistNode *x = zslGetHeader(zsl);
-    for (int i = zslGetHeight(zsl) - 1; i >= 0; i--) {
-        /* stop when we've reached the end of this level or the next node comes
-         * after our target in sorted order. Even though defrag replacements does not impact the skip list order,
-         * when scores are equal, we MUST compare elements lexicographically to maintain correct skip list ordering.
-         * Otherwise we might miss locating the entry. */
-        zskiplistNode *next = x->level[i].forward;
-        while (next &&
-               (next->score < score ||
-                (next->score == score && sdscmp(zslGetNodeElement(next), ele) < 0))) {
-            x = next;
-            next = x->level[i].forward;
-        }
-        update[i] = x;
-    }
-    /* should have arrived at intended node */
-    serverAssert(x->level[0].forward == node);
-
-    zslUpdateNode(zsl, node, newnode, update);
-    *node_ref = newnode; /* update hashtable pointer */
-    allocatorDefragFree(node, allocation_size);
+/* Callback for orderedIndexScanDefrag — when a node is reallocated, update
+ * the hashtable's pointer to it. */
+static void defragZsetNodeCallback(OrderedIndexItem *old_item, OrderedIndexItem *new_item, void *ctx) {
+    hashtable *ht = ctx;
+    bool replaced = hashtableReplaceReallocatedEntry(ht, old_item, new_item);
+    serverAssert(replaced);
+    server.stat_active_defrag_scanned++;
 }
 
 #define DEFRAG_SDS_DICT_NO_VAL 0
@@ -434,15 +386,10 @@ static long scanLaterList(robj *ob, unsigned long *cursor, monotime endtime) {
     return 0;
 }
 
-static void scanLaterZsetCallback(void *privdata, void *element_ref) {
-    activeDefragZsetNode(privdata, element_ref);
-    server.stat_active_defrag_scanned++;
-}
-
 static void scanLaterZset(robj *ob, unsigned long *cursor) {
     serverAssert(ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_SKIPLIST);
     zset *zs = (zset *)objectGetVal(ob);
-    *cursor = hashtableScanDefrag(zs->ht, *cursor, scanLaterZsetCallback, zs->zsl, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
+    *cursor = orderedIndexScanDefrag(zs->oi, *cursor, defragZsetNodeCallback, zs->ht, activeDefragAlloc);
 }
 
 /* Used as hashtable scan callback when all we need is to defrag the hashtable
@@ -482,12 +429,12 @@ static void defragZsetSkiplist(robj *ob) {
     zset *zs = (zset *)objectGetVal(ob);
 
     zset *newzs;
-    zskiplist *newzsl;
     if ((newzs = activeDefragAlloc(zs))) {
         objectSetVal(ob, newzs);
         zs = newzs;
     }
-    if ((newzsl = activeDefragAlloc(zs->zsl))) zs->zsl = newzsl;
+    OrderedIndex *newoi = orderedIndexDefragInternals(zs->oi, activeDefragAlloc);
+    if (newoi) zs->oi = newoi;
 
     hashtable *newtable;
     if ((newtable = hashtableDefragTables(zs->ht, activeDefragAlloc))) zs->ht = newtable;
@@ -497,7 +444,7 @@ static void defragZsetSkiplist(robj *ob) {
     else {
         unsigned long cursor = 0;
         do {
-            cursor = hashtableScanDefrag(zs->ht, cursor, activeDefragZsetNode, zs->zsl, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
+            cursor = orderedIndexScanDefrag(zs->oi, cursor, defragZsetNodeCallback, zs->ht, activeDefragAlloc);
         } while (cursor != 0);
     }
 }

src/geo.c

@@ -29,15 +29,14 @@
  */
 
 #include "geo.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "geohash_helper.h"
 #include "debugmacro.h"
 #include "pqsort.h"
 
 /* Things exported from t_zset.c only for geo.c, since it is the only other
  * part of the server that requires close zset introspection. */
 unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
-int zslValueLteMax(double value, zrangespec *spec);
 
 /* ====================================================================
  * This file implements the following commands:
@@ -295,7 +294,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
             score = zzlGetScore(sptr);
 
             /* If we fell out of range, break. */
-            if (!zslValueLteMax(score, &range)) break;
+            if (!zsetScoreLteMax(score, &range)) break;
 
             vstr = lpGetValue(eptr, &vlen, &vlong);
             if (geoWithinShape(shape, score, xy, &distance) == C_OK) {
@@ -308,26 +307,25 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
         }
     } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
         zset *zs = objectGetVal(zobj);
-        zskiplist *zsl = zs->zsl;
-        zskiplistNode *ln;
+        OrderedIndexIterator iter;
+        orderedIndexInitIterator(&iter, zs->oi);
+        orderedIndexSeekToScoreRange(&iter, range.min, range.max, range.minex, range.maxex, 0);
+        OrderedIndexItem *ln;
 
-        if ((ln = zslNthInRange(zsl, &range, 0, NULL)) == NULL) {
-            /* Nothing exists starting at our min.  No results. */
-            return 0;
-        }
-
-        while (ln) {
+        while ((ln = orderedIndexNext(&iter)) != NULL) {
             double xy[2];
             double distance = 0;
+            double score = orderedIndexGetScore(ln);
             /* Abort when the node is no longer in range. */
-            if (!zslValueLteMax(ln->score, &range)) break;
-            if (geoWithinShape(shape, ln->score, xy, &distance) == C_OK) {
+            if (!zsetScoreLteMax(score, &range)) break;
+            if (geoWithinShape(shape, score, xy, &distance) == C_OK) {
                 /* Append the new element. */
-                sds ele = zslGetNodeElement(ln);
-                geoArrayAppend(ga, xy, distance, ln->score, sdsdup(ele));
+                const char *ele;
+                size_t ele_len;
+                orderedIndexGetElementRaw(ln, &ele, &ele_len);
+                geoArrayAppend(ga, xy, distance, score, sdsnewlen(ele, ele_len));
             }
             if (ga->used && limit && ga->used >= limit) break;
-            ln = ln->level[0].forward;
         }
     }
     return ga->used - origincount;
@@ -833,15 +831,15 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
         }
 
         for (i = 0; i < returned_items; i++) {
-            zskiplistNode *znode;
+            OrderedIndexItem *znode;
             geoPoint *gp = ga.array + i;
             gp->dist /= shape.conversion; /* Fix according to unit. */
             double score = storedist ? gp->dist : gp->score;
             size_t elelen = sdslen(gp->member);
 
             if (maxelelen < elelen) maxelelen = elelen;
             totelelen += elelen;
-            znode = zslInsert(zs->zsl, score, gp->member);
+            znode = orderedIndexInsert(zs->oi, score, gp->member, elelen);
             serverAssert(hashtableAdd(zs->ht, znode));
             sdsfree(gp->member);
             gp->member = NULL;

src/lazyfree.c

@@ -1,5 +1,5 @@
 #include "server.h"
-#include "skiplist.h"
+#include "ordered_index.h"
 #include "bio.h"
 #include "functions.h"
 #include "cluster.h"
@@ -144,7 +144,7 @@ size_t lazyfreeGetFreeEffort(robj *key, robj *obj, int dbid) {
         return hashtableSize(ht);
     } else if (obj->type == OBJ_ZSET && obj->encoding == OBJ_ENCODING_SKIPLIST) {
         zset *zs = objectGetVal(obj);
-        return zslGetLength(zs->zsl);
+        return orderedIndexLength(zs->oi);
     } else if (obj->type == OBJ_HASH && obj->encoding == OBJ_ENCODING_HASHTABLE) {
         hashtable *ht = objectGetVal(obj);
         return hashtableSize(ht);