Reduce large sorted set memory with a packed B+ tree (#15635)

Large sorted sets (beyond `zset-max-listpack-entries` /
`zset-max-listpack-value`) now use a new `btree` encoding instead of
skiplist + dict. The skiplist and the dict keep separate copies of the
same index information. The new encoding uses one packed B+ tree plus a
hash index built for this purpose, and they share information. Small
sets still use listpack. There are no new config options.

### Design

**Score tree.** A B+ tree ordered by (score, member) that owns all
member data. Leaves are single allocations of at most 4095 bytes holding
at most 96 elements, laid out as packed scores, u16 record offsets, one
hash tag byte per member, then member records growing from the end. Each
double is mapped to an order-preserving integer, and a leaf stores a
base value plus bit-packed differences, so dense or integer-like scores
take a few bits each. Members of 320 bytes or more live in their own
allocation; their record keeps the pointer, length and hash, so a score
change does not copy them. Inner nodes store child pointers, the max
score below each child and per-child element counts. The counts make
rank queries a tree descent.

**Member index.** An open-addressing hash table with 8-slot buckets. A
slot holds only an 8-bit hash tag and a 16-bit leaf ID; IDs widen to 32
bits before the 16-bit range runs out. A lookup goes from the tag to the
leaf, then compares members inside the leaf. Each bucket has a small
per-home summary of tags, which rejects most misses without probing.
Resizes are incremental and move whole leaves from the old table to the
new one.

**ZSCAN.** The cursor is a reverse-binary counter over groups of hash
buckets, as in `dictScan()`. It depends only on member hashes, so it
stays valid across resizes. Like SCAN, a member may be returned more
than once.

### Behavior changes
- `OBJECT ENCODING` reports `btree`. Keys are no longer stored as
skiplists; skiplist is only used as a temporary structure inside
set-algebra commands.
- ZSCAN reply order and cursor values differ from the skiplist encoding.
A member can be returned more than once even when no resize happened.
- `MEMORY USAGE` reports the smaller footprint.
- RDB and AOF formats are unchanged. Btree sets are saved as
`RDB_TYPE_ZSET_2`, so old and new versions (including mixed-version
replication) exchange data without migration.

### Implementation
- `src/zset_btree.c`, `src/zset_btree.h`: the new encoding.
- `t_zset.c`: dispatch for every command, plus bulk paths.
Listpack-to-btree conversion and ZRANGESTORE append sorted input
directly into leaves. ZUNIONSTORE builds the result btree directly
instead of a temporary skiplist. ZMSCORE and multi-element pops are
batched.
- `rdb.c`: loads large sets into the btree and validates listpack scores
before conversion.
- Adapted to the new encoding: `aof.c`, `db.c` (ZSCAN), `defrag.c`,
`lazyfree.c`, `object.c` (free, dup, dismiss in fork child), `geo.c`,
`sort.c`, `module.c` (range iteration, `RM_ScanKey`), `debug.c`.
- `dict.c`: `dictFindLinkWithHash()`, so ZUNIONSTORE can reuse a member
hash it already computed.

---------

Co-authored-by: fcostaoliveira <filipe@redis.com>
Co-authored-by: debing.sun <debing.sun@redis.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Moti Cohen <moti.cohen@redis.com>

src/Makefile

@@ -401,7 +401,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
 REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
-REDIS_SERVER_OBJ=threads_mngr.o memory_prefetch.o adlist.o quicklist.o ae.o anet.o dict.o ebuckets.o eventnotifier.o iothread.o mstr.o entry.o kvstore.o fwtree.o estore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o lzf_d_2.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_set_hashtable.o t_set_listpack.o t_set_intset.o t_zset.o t_hash.o bless.o t_array.o sparsearray.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_asm.o cluster_legacy.o cluster_slot_stats.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o bitroar.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o lolwut8.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o logreqres.o keymeta.o chk.o hotkeys.o gcra.o vector.o fast_float_strtod.o client_comp.o
+REDIS_SERVER_OBJ=threads_mngr.o memory_prefetch.o adlist.o quicklist.o ae.o anet.o dict.o ebuckets.o eventnotifier.o iothread.o mstr.o entry.o kvstore.o fwtree.o estore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o lzf_d_2.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_set_hashtable.o t_set_listpack.o t_set_intset.o t_zset.o zset_btree.o t_hash.o bless.o t_array.o sparsearray.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_asm.o cluster_legacy.o cluster_slot_stats.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o bitroar.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o lolwut8.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o logreqres.o keymeta.o chk.o hotkeys.o gcra.o vector.o fast_float_strtod.o client_comp.o
 REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
 REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crccombine.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o cli_commands.o
 REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)

src/aof.c

@@ -12,6 +12,7 @@
 #include "rio.h"
 #include "functions.h"
 #include "cluster_asm.h"
+#include "zset_btree.h"
 
 #include <signal.h>
 #include <fcntl.h>
@@ -2409,6 +2410,27 @@ int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
             items--;
         }
         dictResetIterator(&di);
+    } else if (o->encoding == OBJ_ENCODING_BTREE) {
+        zbtreeIterator iter;
+        const unsigned char *ele;
+        size_t len;
+        double score;
+        zbtreeIteratorStart(o->ptr, 0, &iter);
+        while (zbtreeIteratorNext(&iter, 0, &ele, &len, &score)) {
+            if (count == 0) {
+                int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
+                    AOF_REWRITE_ITEMS_PER_CMD : items;
+                if (!rioWriteBulkCount(r,'*',2+cmd_items*2) ||
+                    !rioWriteBulkString(r,"ZADD",4) ||
+                    !rioWriteBulkObject(r,key))
+                    return 0;
+            }
+            if (!rioWriteBulkDouble(r,score) ||
+                !rioWriteBulkString(r,(char *)ele,len))
+                return 0;
+            if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
     } else {
         serverPanic("Unknown sorted zset encoding");
     }

src/db.c

@@ -27,6 +27,7 @@
 #include <ctype.h>
 #include "bio.h"
 #include "keymeta.h"
+#include "zset_btree.h"
 
 /*-----------------------------------------------------------------------------
  * C-level DB API
@@ -1670,6 +1671,59 @@ typedef struct {
     redisDb *db;  /* database reference for expiration checks */
 } scanData;
 
+typedef struct {
+    client *c;
+    sds pattern;
+    int pattern_len;
+    int use_pattern;
+    unsigned long emitted;
+    int streaming;
+    void *cursor_reply;
+    void *replylen;
+    struct {
+        const unsigned char *ele;
+        size_t len;
+        double score;
+    } buffered[32];
+} zsetBtreeScanData;
+
+static void zsetBtreeScanAddReply(client *c, const unsigned char *ele, size_t len, double score) {
+    addReplyBulkCBuffer(c, ele, len);
+    char buf[MAX_D2STRING_CHARS];
+    int scorelen = d2string(buf, sizeof(buf), score);
+    addReplyBulkCBuffer(c, buf, scorelen);
+}
+
+/* Buffer small replies so their headers need no deferred reply nodes. The
+ * borrowed members remain valid throughout this read-only command. Larger
+ * replies switch to streaming, keeping temporary storage bounded. */
+static void zsetBtreeScanReply(void *privdata, const unsigned char *ele,
+                               size_t len, double score)
+{
+    zsetBtreeScanData *data = privdata;
+    if (data->use_pattern &&
+        !stringmatchlen(data->pattern, data->pattern_len, (char *)ele, len, 0))
+        return;
+    size_t capacity = sizeof(data->buffered) / sizeof(data->buffered[0]);
+    if (!data->streaming && data->emitted / 2 < capacity) {
+        size_t i = data->emitted / 2;
+        data->buffered[i].ele = ele;
+        data->buffered[i].len = len;
+        data->buffered[i].score = score;
+    } else {
+        if (!data->streaming) {
+            data->streaming = 1;
+            data->cursor_reply = addReplyDeferredLen(data->c);
+            data->replylen = addReplyDeferredLen(data->c);
+            for (size_t i = 0; i < capacity; i++)
+                zsetBtreeScanAddReply(data->c, data->buffered[i].ele,
+                                      data->buffered[i].len, data->buffered[i].score);
+        }
+        zsetBtreeScanAddReply(data->c, ele, len, score);
+    }
+    data->emitted += 2;
+}
+
 /* Helper function to compare key type in scan commands */
 int objectTypeCompare(robj *o, long long target) {
     if (o->type != OBJ_MODULE) {
@@ -2015,6 +2069,26 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
         else
             serverAssert(cur_length == array_reply_len); /* fail on corrupt data */
         return;
+    } else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_BTREE) {
+        vecRelease(&keys);
+        addReplyArrayLen(c, 2);
+        zsetBtreeScanData data = {
+            .c = c, .pattern = pat, .pattern_len = patlen, .use_pattern = use_pattern
+        };
+        uint64_t next = zbtreeScan(o->ptr, cursor, count, zsetBtreeScanReply, &data);
+        char cursor_buf[LONG_STR_SIZE];
+        int cursor_len = ull2string(cursor_buf, sizeof(cursor_buf), next);
+        if (data.streaming) {
+            setDeferredReplyBulkSds(c, data.cursor_reply, sdsnewlen(cursor_buf, cursor_len));
+            setDeferredArrayLen(c, data.replylen, data.emitted);
+        } else {
+            addReplyBulkCBuffer(c, cursor_buf, cursor_len);
+            addReplyArrayLen(c, data.emitted);
+            for (size_t i = 0; i < data.emitted / 2; i++)
+                zsetBtreeScanAddReply(c, data.buffered[i].ele,
+                                      data.buffered[i].len, data.buffered[i].score);
+        }
+        return;
     } else if ((o->type == OBJ_HASH || o->type == OBJ_ZSET) &&
                o->encoding == OBJ_ENCODING_LISTPACK)
     {

src/debug.c

@@ -22,6 +22,7 @@
 #include "fpconv_dtoa.h"
 #include "fast_float_strtod.h"
 #include "cluster.h"
+#include "zset_btree.h"
 #include "threads_mngr.h"
 #include "script.h"
 #include "cluster_asm.h"
@@ -341,6 +342,20 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o)
                 xorDigest(digest,eledigest,20);
             }
             dictResetIterator(&di);
+        } else if (o->encoding == OBJ_ENCODING_BTREE) {
+            zbtreeIterator iter;
+            const unsigned char *ele;
+            size_t elelen;
+            double score;
+            zbtreeIteratorStart(o->ptr, 0, &iter);
+            while (zbtreeIteratorNext(&iter, 0, &ele, &elelen, &score)) {
+                const int len = fpconv_dtoa(score, buf);
+                buf[len] = '\0';
+                memset(eledigest,0,20);
+                mixDigest(eledigest,ele,elelen);
+                mixDigest(eledigest,buf,strlen(buf));
+                xorDigest(digest,eledigest,20);
+            }
         } else {
             serverPanic("Unknown sorted set encoding");
         }

src/defrag.c

@@ -19,6 +19,7 @@
  */
 
 #include "server.h"
+#include "zset_btree.h"
 #include "bitroar.h"
 #include <stddef.h>
 #include <math.h>
@@ -116,6 +117,7 @@ typedef struct {
      * before "later" and we will search by key name to find the entry when we defrag the item later. */
     list *defrag_later;
     unsigned long defrag_later_cursor;
+    zbtreeDefragState zset_btree_state;
 } defragKeysCtx;
 static_assert(offsetof(defragKeysCtx, kvstate) == 0, "defragStageKvstoreHelper requires this");
 
@@ -729,6 +731,26 @@ void defragZsetSkiplist(defragKeysCtx *ctx, kvobj *ob) {
         zs->dict = newdict;
 }
 
+/* Defragment a B+ tree now, or schedule its pages for incremental work. */
+void defragZsetBtree(defragKeysCtx *ctx, kvobj *ob) {
+    zbtreeSet *zs = ob->ptr;
+    serverAssert(ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_BTREE);
+
+    ob->ptr = zs = zbtreeDefragStart(zs, activeDefragAlloc);
+    if (zbtreeLength(zs) > server.active_defrag_max_scan_fields) {
+        defragLater(ctx, ob);
+    } else {
+        zbtreeDefragState state = {0};
+        unsigned long scanned;
+        while (!zbtreeDefragStep(zs, &state,
+                                 activeDefragAllocWithoutFree,
+                                 activeDefragFree, &scanned))
+        {
+            server.stat_active_defrag_scanned += scanned;
+        }
+    }
+}
+
 void defragHash(defragKeysCtx *ctx, kvobj *ob) {
     dict *d, *newd;
     serverAssert(ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_HT);
@@ -1218,6 +1240,8 @@ void defragKey(defragKeysCtx *ctx, dictEntry *de, dictEntryLink link) {
                 ob->ptr = newzl;
         } else if (ob->encoding == OBJ_ENCODING_SKIPLIST) {
             defragZsetSkiplist(ctx, ob);
+        } else if (ob->encoding == OBJ_ENCODING_BTREE) {
+            defragZsetBtree(ctx, ob);
         } else {
             serverPanic("Unknown sorted set encoding");
         }
@@ -1358,7 +1382,15 @@ void defragPubsubScanCallback(void *privdata, const dictEntry *de, dictEntryLink
 
 /* returns 0 more work may or may not be needed (see non-zero cursor),
  * and 1 if time is up and more work is needed. */
-int defragLaterItem(kvobj *ob, unsigned long *cursor, monotime endtime, int dbid) {
+int defragLaterItem(defragKeysCtx *ctx, kvobj *ob, monotime endtime) {
+    unsigned long *cursor = &ctx->defrag_later_cursor;
+    zbtreeDefragState *zset_btree_state = &ctx->zset_btree_state;
+    int is_btree = ob && ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_BTREE;
+    if (!is_btree && zset_btree_state->tree) {
+        zbtreeDefragStateReset(zset_btree_state);
+        *cursor = 0;
+    }
+
     if (ob) {
         if (ob->type == OBJ_LIST && ob->encoding == OBJ_ENCODING_QUICKLIST) {
             return scanLaterList(ob, cursor, endtime);
@@ -1366,6 +1398,15 @@ int defragLaterItem(kvobj *ob, unsigned long *cursor, monotime endtime, int dbid
             scanLaterSet(ob, cursor);
         } else if (ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_SKIPLIST) {
             scanLaterZset(ob, cursor);
+        } else if (ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_BTREE) {
+            if (*cursor == 0)
+                zbtreeDefragStateReset(zset_btree_state);
+            unsigned long scanned;
+            int done = zbtreeDefragStep(ob->ptr, zset_btree_state,
+                                        activeDefragAllocWithoutFree,
+                                        activeDefragFree, &scanned);
+            server.stat_active_defrag_scanned += scanned;
+            *cursor = done ? 0 : 1;
         } else if (ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_HT) {
             scanLaterHash(ob, cursor);
         } else if (ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_TMPL_ARRAY) {
@@ -1375,7 +1416,7 @@ int defragLaterItem(kvobj *ob, unsigned long *cursor, monotime endtime, int dbid
         } else if (ob->type == OBJ_MODULE) {
             robj keyobj;
             initStaticStringObject(keyobj, kvobjGetKey(ob));
-            return moduleLateDefrag(&keyobj, ob, cursor, endtime, dbid);
+            return moduleLateDefrag(&keyobj, ob, cursor, endtime, ctx->dbid);
         } else if (ob->type == OBJ_ARRAY) {
             redisArray *ar = ob->ptr;
             *cursor = arDefragIncremental(&ar, *cursor, activeDefragAlloc);
@@ -1384,9 +1425,11 @@ int defragLaterItem(kvobj *ob, unsigned long *cursor, monotime endtime, int dbid
             *cursor = bitroarDefragIncremental(ob, *cursor);
         } else {
             *cursor = 0; /* object type/encoding may have changed since we schedule it for later */
+            zbtreeDefragStateReset(zset_btree_state);
         }
     } else {
         *cursor = 0; /* object may have been deleted already */
+        zbtreeDefragStateReset(zset_btree_state);
     }
     return 0;
 }
@@ -1415,7 +1458,7 @@ static doneStatus defragLaterStep(void *ctx, monotime endtime) {
         long long key_defragged = server.stat_active_defrag_hits;
         if (server.memory_tracking_enabled && kv)
             oldsize = kvobjAllocSize(kv);
-        int timeout = (defragLaterItem(kv, &defrag_keys_ctx->defrag_later_cursor, endtime, defrag_keys_ctx->dbid) == 1);
+        int timeout = (defragLaterItem(defrag_keys_ctx, kv, endtime) == 1);
         if (server.memory_tracking_enabled && kv)
             updateSlotAllocSize(db, slot, kv, oldsize, kvobjAllocSize(kv));
         if (key_defragged != server.stat_active_defrag_hits) {

src/dict.c

@@ -75,7 +75,7 @@ static int _dictInit(dict *d, dictType *type);
 static dictEntryLink dictGetNextLink(dictEntry *de);
 static void dictSetNext(dictEntry *de, dictEntry *next);
 static int dictDefaultCompare(dictCmpCache *cache, const void *key1, const void *key2);
-static dictEntryLink dictFindLinkInternal(dict *d, const void *key, dictEntryLink *bucket);
+static dictEntryLink dictFindLinkInternal(dict *d, const void *key, uint64_t hash, dictEntryLink *bucket);
 dictEntryLink dictFindLinkForInsert(dict *d, const void *key, dictEntry **existing);
 static dictEntry *dictInsertKeyAtLink(dict *d, void *key __stored_key, dictEntryLink link);
 
@@ -763,20 +763,19 @@ void dictRelease(dict *d)
  * 
  * bucket - return pointer to bucket that the key was mapped. unless dict is empty.
  */
-static dictEntryLink dictFindLinkInternal(dict *d, const void *key, dictEntryLink *bucket) {
+static dictEntryLink dictFindLinkInternal(dict *d, const void *key, uint64_t hash, dictEntryLink *bucket) {
     dictCmpCache cmpCache = {0};
     dictEntryLink link;
     uint64_t idx;
     int table;
-    
+
     if (bucket) {
         *bucket = NULL;
     } else {
         /* If dict is empty and no need to find bucket, return NULL */
-        if (dictSize(d) == 0) return NULL; 
+        if (dictSize(d) == 0) return NULL;
     }
 
-    const uint64_t hash = dictGetHash(d, key);
     idx = hash & DICTHT_SIZE_MASK(d->ht_size_exp[0]);
     keyCmpFunc cmpFunc = dictGetCmpFunc(d);
 
@@ -865,10 +864,23 @@ dictEntryLink dictFindLink(dict *d, const void *key, dictEntryLink *bucket) {
     if (unlikely(dictSize(d) == 0))
         return NULL;
     
-    return dictFindLinkInternal(d, key, bucket);
+    return dictFindLinkInternal(d, key, dictGetHash(d, key), bucket);
 }
 
-/* Set the key with link 
+/* Same as dictFindLink(), but for a caller that already computed this exact
+ * key's hash (via dictGetHash(d, key)) for another purpose and wants to avoid
+ * paying for the dict type's hashFunction() a second time. The hash is not
+ * validated against key -- passing a mismatched hash silently searches the
+ * wrong bucket. */
+dictEntryLink dictFindLinkWithHash(dict *d, const void *key, uint64_t hash, dictEntryLink *bucket) {
+    if (bucket) *bucket = NULL;
+    if (unlikely(dictSize(d) == 0))
+        return NULL;
+
+    return dictFindLinkInternal(d, key, hash, bucket);
+}
+
+/* Set the key with link
  *
  * link:    - When `newItem` is set, `link` points to the bucket of the key.
  *          - When `newItem` is not set, `link` points to the link of the key.
@@ -893,7 +905,8 @@ void dictSetKeyAtLink(dict *d, void *key __stored_key, dictEntryLink *link, int
         if (snap[0] != d->ht_size_exp[0] || snap[1] != d->ht_size_exp[1] || *link == NULL) {
             dictEntryLink bucket;
             /* Bypass dictFindLink() to search bucket even if dict is empty!!! */
-            *link = dictFindLinkInternal(d, dictStoredKey2Key(d, key), &bucket);
+            const void *lookup_key = dictStoredKey2Key(d, key);
+            *link = dictFindLinkInternal(d, lookup_key, dictGetHash(d, lookup_key), &bucket);
             assert(bucket != NULL);
             assert(*link == NULL);
             *link = bucket; /* On newItem the link should be the bucket */

src/dict.h

@@ -304,6 +304,7 @@ void dictCombineStats(dictStats *from, dictStats *into);
 void dictFreeStats(dictStats *stats);
 
 dictEntryLink dictFindLink(dict *d, const void *key, dictEntryLink *bucket);
+dictEntryLink dictFindLinkWithHash(dict *d, const void *key, uint64_t hash, dictEntryLink *bucket);
 void dictSetKeyAtLink(dict *d, void *key __stored_key, dictEntryLink *link, int newItem);
 
 /* API relevant only when dict is used as a hash-map (no_value=0) */ 

src/geo.c

@@ -30,6 +30,7 @@
 
 #include "geo.h"
 #include "geohash_helper.h"
+#include "zset_btree.h"
 #include "debugmacro.h"
 #include "pqsort.h"
 
@@ -319,6 +320,21 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
             if (ga->used && limit && ga->used >= limit) break;
             ln = ln->level[0].forward;
         }
+    } else if (zobj->encoding == OBJ_ENCODING_BTREE) {
+        zbtreeIterator iter;
+        const unsigned char *ele;
+        size_t len;
+        double score;
+        if (!zbtreeIteratorSeekScore(zobj->ptr, range.min, range.minex, 0, &iter, NULL))
+            return 0;
+        while (zbtreeIteratorNext(&iter, 0, &ele, &len, &score)) {
+            double xy[2];
+            double distance = 0;
+            if (!zslValueLteMax(score, &range)) break;
+            if (geoWithinShape(shape, score, xy, &distance) == C_OK)
+                geoArrayAppend(ga, xy, distance, score, sdsnewlen(ele, len));
+            if (ga->used && limit && ga->used >= limit) break;
+        }
     }
     return ga->used - origincount;
 }
@@ -804,32 +820,23 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
     } else {
         /* Target key, create a sorted set with the results. */
         robj *zobj;
-        zset *zs;
         int i;
-        size_t maxelelen = 0, totelelen = 0;
 
-        if (returned_items) {
-            zobj = createZsetObject();
-            zs = zobj->ptr;
-        }
+        if (returned_items)
+            zobj = zsetTypeCreate(returned_items, 0);
 
         for (i = 0; i < returned_items; i++) {
-            zskiplistNode *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);
-            serverAssert(dictAdd(zs->dict, znode, NULL) == DICT_OK);
-            sdsfree(gp->member); /* zslInsert copies the sds, so free the original */
+            int out_flags;
+            serverAssert(zsetAdd(zobj, score, gp->member, ZADD_IN_NONE, &out_flags, NULL));
+            serverAssert(out_flags & ZADD_OUT_ADDED);
+            sdsfree(gp->member);
             gp->member = NULL;
         }
 
         if (returned_items) {
-            zsetConvertToListpackIfNeeded(zobj,maxelelen,totelelen);
             setKey(c,c->db,storekey,&zobj,0);
             notifyKeyspaceEvent(NOTIFY_ZSET,flags & GEOSEARCH ? "geosearchstore" : "georadiusstore",storekey,
                                 c->db->id);

src/lazyfree.c

@@ -155,6 +155,8 @@ size_t lazyfreeGetFreeEffort(robj *key, robj *obj, int dbid) {
     } else if (obj->type == OBJ_ZSET && obj->encoding == OBJ_ENCODING_SKIPLIST){
         zset *zs = obj->ptr;
         return zs->zsl->length;
+    } else if (obj->type == OBJ_ZSET && obj->encoding == OBJ_ENCODING_BTREE) {
+        return zsetLength(obj);
     } else if (obj->type == OBJ_HASH && obj->encoding == OBJ_ENCODING_HT) {
         dict *ht = obj->ptr;
         return dictSize(ht);

src/module.c

@@ -46,6 +46,7 @@
 #include "call_reply.h"
 #include "hdr_histogram.h"
 #include "crc16_slottable.h"
+#include "zset_btree.h"
 #include <dlfcn.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
@@ -198,7 +199,7 @@ struct RedisModuleKey {
             zlexrangespec lrs;     /* Lex range. */
             uint32_t start;        /* Start pos for positional ranges. */
             uint32_t end;          /* End pos for positional ranges. */
-            void *current;         /* Zset iterator current node. */
+            void *current;         /* Node, or B+ tree rank plus one. */
             int er;                /* Zset iterator end reached flag
                                        (true if end was reached). */
         } zset;
@@ -5455,6 +5456,18 @@ void zsetKeyReset(RedisModuleKey *key) {
     key->u.zset.er = 1;
 }
 
+/* Read one B+ tree member by rank. Seeking for every call keeps no leaf
+ * pointer across module code, which may modify the sorted set. */
+static int moduleZsetBtreeElementAtRank(RedisModuleKey *key,
+                                        unsigned long rank,
+                                        const unsigned char **ele,
+                                        size_t *len, double *score)
+{
+    zbtreeIterator iter;
+    if (!zbtreeIteratorSeekRank(key->kv->ptr, rank, &iter)) return 0;
+    return zbtreeIteratorNext(&iter, 0, ele, len, score);
+}
+
 /* Stop a sorted set iteration. */
 void RM_ZsetRangeStop(RedisModuleKey *key) {
     if (!key->kv || key->kv->type != OBJ_ZSET) return;
@@ -5502,6 +5515,19 @@ int zsetInitScoreRange(RedisModuleKey *key, double min, double max, int minex, i
         zskiplist *zsl = zs->zsl;
         key->u.zset.current = first ? zslNthInRange(zsl, zrs, 0, NULL) :
                                       zslNthInRange(zsl, zrs, -1, NULL);
+    } else if (key->kv->encoding == OBJ_ENCODING_BTREE) {
+        zbtreeIterator iter;
+        unsigned long rank;
+        double score;
+        int found = first ?
+            zbtreeIteratorSeekScore(key->kv->ptr, min, minex, 0, &iter, &rank) :
+            zbtreeIteratorSeekScore(key->kv->ptr, max, maxex, 1, &iter, &rank);
+        if (found) {
+            serverAssert(moduleZsetBtreeElementAtRank(key, rank, NULL, NULL, &score));
+            /* Module callers can pass NaN bounds, so check both sides. */
+            if (zslValueGteMin(score, zrs) && zslValueLteMax(score, zrs))
+                key->u.zset.current = (void *)(uintptr_t)(rank + 1);
+        }
     } else {
         serverPanic("Unsupported zset encoding");
     }
@@ -5566,6 +5592,23 @@ int zsetInitLexRange(RedisModuleKey *key, RedisModuleString *min, RedisModuleStr
         zskiplist *zsl = zs->zsl;
         key->u.zset.current = first ? zslNthInLexRange(zsl,zlrs,0,NULL) :
                                       zslNthInLexRange(zsl,zlrs,-1,NULL);
+    } else if (key->kv->encoding == OBJ_ENCODING_BTREE) {
+        zbtreeIterator iter;
+        unsigned long rank;
+        int found = first ?
+            zbtreeIteratorSeekLex(key->kv->ptr, zlrs->min, zlrs->minex, 0, &iter, &rank) :
+            zbtreeIteratorSeekLex(key->kv->ptr, zlrs->max, zlrs->maxex, 1, &iter, &rank);
+        if (found) {
+            const unsigned char *ele;
+            size_t len;
+            serverAssert(moduleZsetBtreeElementAtRank(key, rank, &ele, &len, NULL));
+            sds value = sdsnewlen(ele, len);
+            int in_range = zslLexValueGteMin(value, zlrs) &&
+                           zslLexValueLteMax(value, zlrs);
+            sdsfree(value);
+            if (in_range)
+                key->u.zset.current = (void *)(uintptr_t)(rank + 1);
+        }
     } else {
         serverPanic("Unsupported zset encoding");
     }
@@ -5618,6 +5661,17 @@ RedisModuleString *RM_ZsetRangeCurrentElement(RedisModuleKey *key, double *score
         if (score) *score = ln->score;
         sds ele = zslGetNodeElement(ln);
         str = createStringObject(ele,sdslen(ele));
+    } else if (key->kv->encoding == OBJ_ENCODING_BTREE) {
+        const unsigned char *ele;
+        size_t len;
+        double value;
+        unsigned long rank = (uintptr_t)key->u.zset.current - 1;
+        if (!moduleZsetBtreeElementAtRank(key, rank, &ele, &len, &value)) {
+            key->u.zset.er = 1;
+            return NULL;
+        }
+        if (score) *score = value;
+        str = createStringObject((char *)ele, len);
     } else {
         serverPanic("Unsupported zset encoding");
     }
@@ -5684,9 +5738,31 @@ int RM_ZsetRangeNext(RedisModuleKey *key) {
             key->u.zset.current = next;
             return 1;
         }
+    } else if (key->kv->encoding == OBJ_ENCODING_BTREE) {
+        unsigned long rank = (uintptr_t)key->u.zset.current - 1;
+        if (++rank >= zbtreeLength(key->kv->ptr)) goto end;
+        const unsigned char *ele;
+        size_t len;
+        double score;
+        if (!moduleZsetBtreeElementAtRank(key, rank, &ele, &len, &score))
+            goto end;
+        if (key->u.zset.type == REDISMODULE_ZSET_RANGE_SCORE) {
+            if (!zslValueLteMax(score, &key->u.zset.rs)) goto end;
+        } else if (key->u.zset.type == REDISMODULE_ZSET_RANGE_LEX) {
+            sds value = sdsnewlen(ele, len);
+            int valid = zslLexValueLteMax(value, &key->u.zset.lrs);
+            sdsfree(value);
+            if (!valid) goto end;
+        }
+        key->u.zset.current = (void *)(uintptr_t)(rank + 1);
+        return 1;
     } else {
         serverPanic("Unsupported zset encoding");
     }
+
+end:
+    key->u.zset.er = 1;
+    return 0;
 }
 
 /* Go to the previous element of the sorted set iterator. Returns 1 if there was
@@ -5748,9 +5824,32 @@ int RM_ZsetRangePrev(RedisModuleKey *key) {
             key->u.zset.current = prev;
             return 1;
         }
+    } else if (key->kv->encoding == OBJ_ENCODING_BTREE) {
+        unsigned long rank = (uintptr_t)key->u.zset.current - 1;
+        if (rank == 0) goto end;
+        rank--;
+        const unsigned char *ele;
+        size_t len;
+        double score;
+        if (!moduleZsetBtreeElementAtRank(key, rank, &ele, &len, &score))
+            goto end;
+        if (key->u.zset.type == REDISMODULE_ZSET_RANGE_SCORE) {
+            if (!zslValueGteMin(score, &key->u.zset.rs)) goto end;
+        } else if (key->u.zset.type == REDISMODULE_ZSET_RANGE_LEX) {
+            sds value = sdsnewlen(ele, len);
+            int valid = zslLexValueGteMin(value, &key->u.zset.lrs);
+            sdsfree(value);
+            if (!valid) goto end;
+        }
+        key->u.zset.current = (void *)(uintptr_t)(rank + 1);
+        return 1;
     } else {
         serverPanic("Unsupported zset encoding");
     }
+
+end:
+    key->u.zset.er = 1;
+    return 0;
 }
 
 /* --------------------------------------------------------------------------
@@ -12486,6 +12585,34 @@ typedef struct {
     RedisModuleScanKeyCB fn;
 } ScanKeyCBData;
 
+typedef struct {
+    robj *field;
+    double score;
+} BtreeScanItem;
+
+typedef struct {
+    BtreeScanItem *items;
+    size_t count;
+    size_t capacity;
+} BtreeScanData;
+
+/* B+ tree leaves may move if module code changes the sorted set. Collect one
+ * small batch before calling the module so no tree pointer survives a
+ * callback. */
+static void moduleBtreeScanCollect(void *privdata, const unsigned char *ele,
+                                   size_t len, double score)
+{
+    BtreeScanData *data = privdata;
+    if (data->count == data->capacity) {
+        data->capacity = data->capacity ? data->capacity * 2 : 64;
+        data->items = zrealloc(data->items, data->capacity * sizeof(*data->items));
+    }
+
+    data->items[data->count].field = createStringObject((char *)ele, len);
+    data->items[data->count].score = score;
+    data->count++;
+}
+
 static void moduleScanKeyCallback(void *privdata, const dictEntry *de, dictEntryLink plink) {
     UNUSED(plink);
     ScanKeyCBData *data = privdata;
@@ -12636,6 +12763,34 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
         cursor->cursor = 1;
         cursor->done = 1;
         ret = 0;
+    } else if (kv->type == OBJ_ZSET && kv->encoding == OBJ_ENCODING_BTREE) {
+        BtreeScanData data = {0};
+        cursor->cursor = zbtreeScan(kv->ptr, cursor->cursor, 64, moduleBtreeScanCollect, &data);
+
+        size_t i = 0;
+        for (; i < data.count; i++) {
+            robj *value = createStringObjectFromLongDouble(data.items[i].score, 0);
+            fn(key, data.items[i].field, value, privdata);
+            decrRefCount(value);
+            decrRefCount(data.items[i].field);
+
+            /* Removing the current member is safe. If the callback removes the
+             * key or changes its encoding, this B+ tree cursor has no meaning. */
+            if (key->kv != kv || kv->encoding != OBJ_ENCODING_BTREE) {
+                cursor->cursor = 0;
+                cursor->done = 1;
+                ret = 0;
+                i++;
+                break;
+            }
+        }
+        while (i < data.count) decrRefCount(data.items[i++].field);
+        zfree(data.items);
+
+        if (cursor->cursor == 0) {
+            cursor->done = 1;
+            ret = 0;
+        }
     } else if (kv->type == OBJ_ZSET || kv->type == OBJ_HASH) {
         unsigned char *lp, *p;
         /* is hash with expiry on fields, then lp tuples are [field][value][expire] */

src/object.c

@@ -16,6 +16,7 @@
 #include "functions.h"
 #include "intset.h"  /* Compact integer set structure */
 #include "cluster_asm.h"
+#include "zset_btree.h"
 #include <math.h>
 #include <ctype.h>
 
@@ -510,6 +511,13 @@ robj *createZsetObject(void) {
     return o;
 }
 
+/* Create an empty sorted set using the packed B+ tree encoding. */
+robj *createZsetBtreeObject(void) {
+    robj *o = createObject(OBJ_ZSET, zbtreeCreate());
+    o->encoding = OBJ_ENCODING_BTREE;
+    return o;
+}
+
 robj *createZsetListpackObject(void) {
     unsigned char *lp = lpNew(0);
     robj *o = createObject(OBJ_ZSET,lp);
@@ -586,6 +594,9 @@ void freeZsetObject(robj *o) {
         zslFree(zs->zsl);
         zfree(zs);
         break;
+    case OBJ_ENCODING_BTREE:
+        zbtreeFree(o->ptr);
+        break;
     case OBJ_ENCODING_LISTPACK:
         zfree(o->ptr);
         break;
@@ -769,6 +780,8 @@ void dismissZsetObject(robj *o, size_t size_hint) {
 
         /* Dismiss hash table memory. */
         dismissDictBucketsMemory(zs->dict);
+    } else if (o->encoding == OBJ_ENCODING_BTREE) {
+        zbtreeDismissMemory(o->ptr);
     } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
         dismissMemory(o->ptr, lpBytes((unsigned char*)o->ptr));
     } else {
@@ -1322,6 +1335,7 @@ char *strEncoding(int encoding) {
     case OBJ_ENCODING_TMPL_LP: return "template-listpack";
     case OBJ_ENCODING_TMPL_ARRAY: return "template-array";
     case OBJ_ENCODING_BITMAP_ROARING: return "bitmap-roaring";
+    case OBJ_ENCODING_BTREE: return "btree";
     default: return "unknown";
     }
 }

src/object.h

@@ -78,7 +78,7 @@ struct RedisModuleType;
 #define OBJ_ENCODING_HT 2      /* Encoded as hash table */
 #define OBJ_ENCODING_ZIPMAP 3  /* No longer used: old hash encoding. */
 #define OBJ_ENCODING_LINKEDLIST 4 /* No longer used: old list encoding. */
-#define OBJ_ENCODING_ZIPLIST 5 /* No longer used: old list/hash/zset encoding. */
+#define OBJ_ENCODING_BTREE 5   /* Sorted set as a packed B+ tree */
 #define OBJ_ENCODING_INTSET 6  /* Encoded as intset */
 #define OBJ_ENCODING_SKIPLIST 7  /* Encoded as skiplist */
 #define OBJ_ENCODING_EMBSTR 8  /* Embedded sds string encoding */
@@ -180,6 +180,7 @@ robj *createIntsetObject(void);
 robj *createSetListpackObject(void);
 robj *createHashObject(void);
 robj *createZsetObject(void);
+robj *createZsetBtreeObject(void);
 robj *createZsetListpackObject(void);
 robj *createStreamObject(void);
 robj *createGCRAObject(long long value);

src/rdb.c

@@ -24,6 +24,7 @@
 #include "bio.h"
 #include "cluster_asm.h"
 #include "keymeta.h"
+#include "zset_btree.h"
 
 #include <math.h>
 #include <fcntl.h>
@@ -713,7 +714,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
     case OBJ_ZSET:
         if (o->encoding == OBJ_ENCODING_LISTPACK)
             return rdbSaveType(rdb,RDB_TYPE_ZSET_LISTPACK);
-        else if (o->encoding == OBJ_ENCODING_SKIPLIST)
+        else if (o->encoding == OBJ_ENCODING_SKIPLIST ||
+                 o->encoding == OBJ_ENCODING_BTREE)
             return rdbSaveType(rdb,RDB_TYPE_ZSET_2);
         else
             serverPanic("Unknown sorted set encoding");
@@ -1283,7 +1285,9 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
             serverPanic("Unknown set encoding");
         }
     } else if (o->type == OBJ_ZSET) {
-        /* Save a sorted set value */
+        /* Save a sorted set value. SKIPLIST is only a temporary encoding,
+         * keyspace zsets are LISTPACK or BTREE. */
+        debugServerAssert(o->encoding != OBJ_ENCODING_SKIPLIST);
         if (o->encoding == OBJ_ENCODING_LISTPACK) {
             size_t l = lpBytes((unsigned char*)o->ptr);
 
@@ -1316,6 +1320,24 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
                 nwritten += n;
                 zn = zn->backward;
             }
+        } else if (o->encoding == OBJ_ENCODING_BTREE) {
+            zbtreeIterator iter;
+            const unsigned char *ele;
+            size_t len;
+            double score;
+
+            if ((n = rdbSaveLen(rdb, zbtreeLength(o->ptr))) == -1)
+                return -1;
+            nwritten += n;
+            zbtreeIteratorStart(o->ptr, 1, &iter);
+            while (zbtreeIteratorNext(&iter, 1, &ele, &len, &score)) {
+                if ((n = rdbSaveRawString(rdb, (unsigned char *)ele, len)) == -1)
+                    return -1;
+                nwritten += n;
+                if ((n = rdbSaveBinaryDoubleValue(rdb, score)) == -1)
+                    return -1;
+                nwritten += n;
+            }
         } else {
             serverPanic("Unknown sorted set encoding");
         }
@@ -2543,23 +2565,27 @@ static int _listZiplistEntryConvertAndValidate(unsigned char *p, unsigned int he
     return 1;
 }
 
-/* callback for to check the listpack doesn't have duplicate records */
-static int _lpEntryValidation(unsigned char *p, unsigned int head_count, void *userdata) {
-    struct {
-        int tuple_len;
-        long count;
-        dict *fields;
-        long long last_expireat;
-    } *data = userdata;
+typedef struct {
+    int tuple_len;
+    long count;
+    dict *fields;
+    long long last_expireat;
+    int check_duplicates;
+    int validate_scores;
+} lpValidationData;
 
-    if (data->fields == NULL) {
+/* Validate tuple fields after lpValidateIntegrity checked the entry. */
+static int _lpEntryValidation(unsigned char *p, unsigned int head_count, void *userdata) {
+    lpValidationData *data = userdata;
+
+    if (data->check_duplicates && data->fields == NULL) {
         data->fields = dictCreate(&hashDictType);
         dictExpand(data->fields, head_count/data->tuple_len);
     }
 
     /* If we're checking pairs, then even records are field names. Otherwise
      * we're checking all elements. Add to dict and check that's not a dup */
-    if (data->count % data->tuple_len == 0) {
+    if (data->check_duplicates && data->count % data->tuple_len == 0) {
         unsigned char *str;
         int64_t slen;
         unsigned char buf[LP_INTBUF_SIZE];
@@ -2573,6 +2599,17 @@ static int _lpEntryValidation(unsigned char *p, unsigned int head_count, void *u
         }
     }
 
+    /* A zset's second entry is a score, not an arbitrary hash value. Parse
+     * it before any conversion can store an invalid or NaN score. Integer
+     * entries are always valid, and infinity remains a supported score. */
+    if (data->validate_scores && data->count % data->tuple_len == 1) {
+        unsigned int len;
+        long long integer;
+        unsigned char *str = lpGetValue(p, &len, &integer);
+        double score;
+        if (str && !string2d((char *)str, len, &score)) return 0;
+    }
+
     /* Validate TTL field, only for listpackex. */
     if (data->count % data->tuple_len == 2) {
         long long expire_at;
@@ -2596,17 +2633,15 @@ static int _lpEntryValidation(unsigned char *p, unsigned int head_count, void *u
  * tuple_len indicates what is a logical entry tuple size.
  * Whether tuple is of size 1 (set), 2 (field-value) or 3 (field-value[-ttl]),
  * first element in the tuple must be unique */
-int lpValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep, int tuple_len) {
-    if (!deep)
+static int lpValidateIntegrityAndDupsEx(unsigned char *lp, size_t size,
+                                        int deep, int tuple_len, int validate_scores)
+{
+    if (!deep && !validate_scores)
         return lpValidateIntegrity(lp, size, 0, NULL, NULL);
 
-    /* Keep track of the field names to locate duplicate ones */
-    struct {
-        int tuple_len;
-        long count;
-        dict *fields; /* Initialisation at the first callback. */
-        long long last_expireat; /* Last field's expiry time to ensure order in TTL fields. */
-    } data = {tuple_len, 0, NULL, -1};
+    /* Score validation must visit every entry even when duplicate checking
+     * is disabled. The integrity walker bounds each entry before decoding. */
+    lpValidationData data = {tuple_len, 0, NULL, -1, deep, validate_scores};
 
     int ret = lpValidateIntegrity(lp, size, 1, _lpEntryValidation, &data);
 
@@ -2618,6 +2653,10 @@ int lpValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep, int tup
     return ret;
 }
 
+int lpValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep, int tuple_len) {
+    return lpValidateIntegrityAndDupsEx(lp, size, deep, tuple_len, 0);
+}
+
 /* ---- shared SDS-array helpers ---- */
 
 /* Free 'count' SDS strings and the array holding them. Safe on NULL. */
@@ -3116,16 +3155,16 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
         /* Read sorted set value. */
         uint64_t zsetlen;
         size_t maxelelen = 0, totelelen = 0;
-        zset *zs;
 
         if ((zsetlen = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
         if (zsetlen == 0) goto emptykey;
-
-        o = createZsetObject();
-        zs = o->ptr;
-
-        if (zsetlen > DICT_HT_INITIAL_SIZE && dictTryExpand(zs->dict,zsetlen) != DICT_OK) {
-            rdbReportCorruptRDB("OOM in dictTryExpand %llu", (unsigned long long)zsetlen);
+        /* Decide listpack eligibility only after seeing every member. A
+         * temporary listpack would normalize -0 before a later long member
+         * promotes the object, changing the final B+tree's score bits. */
+        o = createZsetBtreeObject();
+        if (!zbtreeTryReserve(o->ptr, zsetlen)) {
+            rdbReportCorruptRDB("Unable to reserve zset index for %llu elements",
+                                (unsigned long long)zsetlen);
             decrRefCount(o);
             return NULL;
         }
@@ -3134,7 +3173,6 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
         while(zsetlen--) {
             sds sdsele;
             double score;
-            zskiplistNode *znode;
 
             if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL)) == NULL) {
                 decrRefCount(o);
@@ -3162,27 +3200,23 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
                 return NULL;
             }
 
-            /* Don't care about integer-encoded strings. */
-            if (sdslen(sdsele) > maxelelen) maxelelen = sdslen(sdsele);
-            totelelen += sdslen(sdsele);
-
-            znode = zslInsert(zs->zsl,score,sdsele);
-            if (dictAdd(zs->dict, znode, NULL) != DICT_OK) {
+            int out_flags;
+            serverAssert(zsetAdd(o, score, sdsele, ZADD_IN_NX, &out_flags, NULL));
+            if (!(out_flags & ZADD_OUT_ADDED)) {
                 rdbReportCorruptRDB("Duplicate zset fields detected");
                 decrRefCount(o);
-                sdsfree(sdsele); /* zslInsert copies the sds, so we need to free the original */
+                sdsfree(sdsele);
                 return NULL;
             }
-            sdsfree(sdsele); /* zslInsert copies the sds into the node, so free the original */
-        }
-
-        /* Convert *after* loading, since sorted sets are not stored ordered. */
-        if (zsetLength(o) <= server.zset_max_listpack_entries &&
-            maxelelen <= server.zset_max_listpack_value &&
-            lpSafeToAdd(NULL, totelelen))
-        {
-            zsetConvert(o, OBJ_ENCODING_LISTPACK);
+            size_t elelen = sdslen(sdsele);
+            if (elelen > maxelelen) maxelelen = elelen;
+            if (totelelen <= SIZE_MAX - elelen)
+                totelelen += elelen;
+            else
+                totelelen = SIZE_MAX;
+            sdsfree(sdsele);
         }
+        zsetConvertAfterBulkInsert(o, maxelelen, totelelen);
     } else if (rdbtype == RDB_TYPE_HASH) {
         uint64_t len, original_len;
         int ret;
@@ -3820,7 +3854,12 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
             case RDB_TYPE_ZSET_ZIPLIST:
                 {
                     unsigned char *lp = lpNew(encoded_len);
-                    if (!ziplistPairsConvertAndValidateIntegrity(encoded, encoded_len, &lp)) {
+                    int valid = ziplistPairsConvertAndValidateIntegrity(encoded, encoded_len, &lp);
+                    /* See RDB_TYPE_ZSET_LISTPACK below for when scores are parsed. */
+                    if (valid && (deep_integrity_validation ||
+                                  lpLength(lp) / 2 > server.zset_max_listpack_entries))
+                        valid = lpValidateIntegrityAndDupsEx(lp, lpBytes(lp), 0, 2, 1);
+                    if (!valid) {
                         rdbReportCorruptRDB("Zset ziplist integrity check failed.");
                         zfree(lp);
                         zfree(encoded);
@@ -3839,14 +3878,16 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
                     }
 
                     if (zsetLength(o) > server.zset_max_listpack_entries)
-                        zsetConvert(o, OBJ_ENCODING_SKIPLIST);
+                        zsetConvert(o, OBJ_ENCODING_BTREE);
                     else
                         o->ptr = lpShrinkToFit(o->ptr);
                     break;
                 }
             case RDB_TYPE_ZSET_LISTPACK:
                 if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
-                if (!lpValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation, 2)) {
+                if (!lpValidateIntegrityAndDupsEx(encoded, encoded_len,
+                                                  deep_integrity_validation, 2,
+                                                  deep_integrity_validation)) {
                     rdbReportCorruptRDB("Zset listpack integrity check failed.");
                     zfree(encoded);
                     o->ptr = NULL;
@@ -3860,8 +3901,21 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
                     goto emptykey;
                 }
 
-                if (zsetLength(o) > server.zset_max_listpack_entries)
-                    zsetConvert(o, OBJ_ENCODING_SKIPLIST);
+                if (zsetLength(o) > server.zset_max_listpack_entries) {
+                    /* Parsing every score costs a full walk, so a listpack
+                     * that stays one is checked only by deep sanitization.
+                     * A conversion parses every score anyway and must not
+                     * store an invalid one, so check them first. */
+                    if (!deep_integrity_validation &&
+                        !lpValidateIntegrityAndDupsEx(encoded, encoded_len, 0, 2, 1)) {
+                        rdbReportCorruptRDB("Zset listpack integrity check failed.");
+                        zfree(encoded);
+                        o->ptr = NULL;
+                        decrRefCount(o);
+                        return NULL;
+                    }
+                    zsetConvert(o, OBJ_ENCODING_BTREE);
+                }
                 break;
             case RDB_TYPE_HASH_ZIPLIST:
                 {

src/server.c

@@ -8338,6 +8338,7 @@ int __test_num = 0;
 typedef int redisTestProc(int argc, char **argv, int flags);
 int bitopsTest(int argc, char **argv, int flags);
 int zsetTest(int argc, char **argv, int flags);
+int zsetBtreeTest(int argc, char **argv, int flags);
 int vectorTest(int argc, char **argv, int flags);
 int lzfTest(int argc, char **argv, int flags);
 struct redisTest {
@@ -8367,6 +8368,7 @@ struct redisTest {
     {"bitmap", bitopsTest},
     {"rax", raxTest},
     {"zset", zsetTest},
+    {"zsetbtree", zsetBtreeTest},
     {"topk", chkTopKTest},
     {"fastfloat", fastFloatTest},
     {"lzf", lzfTest},

src/server.h

@@ -3895,8 +3895,10 @@ unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
 unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range);
 unsigned long zsetLength(const robj *zobj);
 size_t zsetAllocSize(const robj *o);
+robj *zsetTypeCreate(size_t size_hint, size_t val_len_hint);
 void zsetConvert(robj *zobj, int encoding);
-void zsetConvertToListpackIfNeeded(robj *zobj, size_t maxelelen, size_t totelelen);
+void zsetConvertAndExpand(robj *zobj, int encoding, unsigned long cap);
+void zsetConvertAfterBulkInsert(robj *zobj, size_t maxelelen, size_t totelelen);
 int zsetScore(robj *zobj, sds member, double *score);
 unsigned long zslGetRank(zskiplist *zsl, double score, sds o);
 int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore);

src/sort.c

@@ -10,6 +10,7 @@
 
 #include "fast_float_strtod.h"
 #include "server.h"
+#include "zset_btree.h"
 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
 #include <math.h> /* isnan() */
 #include "cluster.h"
@@ -338,11 +339,11 @@ void sortCommandGeneric(client *c, int readonly) {
         sortby = NULL;
     }
 
-    /* Destructively convert encoded sorted sets for SORT. */
-    if (sortval->type == OBJ_ZSET) {
+    /* Destructively convert listpack encoded sorted sets for SORT. */
+    if (sortval->type == OBJ_ZSET && sortval->encoding == OBJ_ENCODING_LISTPACK) {
         if (server.memory_tracking_enabled)
             oldsize = kvobjAllocSize(sortval);
-        zsetConvert(sortval, OBJ_ENCODING_SKIPLIST);
+        zsetConvert(sortval, OBJ_ENCODING_BTREE);
         if (server.memory_tracking_enabled)
             updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), sortval, oldsize, kvobjAllocSize(sortval));
     }
@@ -351,7 +352,7 @@ void sortCommandGeneric(client *c, int readonly) {
     switch(sortval->type) {
     case OBJ_LIST: vectorlen = listTypeLength(sortval); break;
     case OBJ_SET: vectorlen =  setTypeSize(sortval); break;
-    case OBJ_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
+    case OBJ_ZSET: vectorlen = zsetLength(sortval); break;
     default: vectorlen = 0; serverPanic("Bad SORT type"); /* Avoid GCC warning */
     }
 
@@ -446,54 +447,79 @@ void sortCommandGeneric(client *c, int readonly) {
          * Note that in this case we also handle LIMIT here in a direct
          * way, just getting the required range, as an optimization. */
 
-        zset *zs = sortval->ptr;
-        zskiplist *zsl = zs->zsl;
-        zskiplistNode *ln;
-        sds sdsele;
         int rangelen = vectorlen;
-
-        /* Check if starting point is trivial, before doing log(N) lookup. */
-        if (desc) {
-            long zsetlen = dictSize(((zset*)sortval->ptr)->dict);
-
-            ln = zsl->tail;
-            if (start > 0)
-                ln = zslGetElementByRank(zsl,zsetlen-start);
+        if (sortval->encoding == OBJ_ENCODING_BTREE) {
+            serverAssert(zsetLength(sortval) > 0);
+            zbtreeIterator iter;
+            const unsigned char *ele;
+            size_t len;
+            unsigned long rank =
+                desc ? zsetLength(sortval) - (unsigned long)start - 1 :
+                       (unsigned long)start;
+            serverAssert(zbtreeIteratorSeekRank(sortval->ptr, rank, &iter));
+            while(rangelen--) {
+                serverAssert(zbtreeIteratorNext(&iter, desc, &ele, &len, NULL));
+                vector[j].obj = createStringObject((char *)ele,len);
+                vector[j].u.score = 0;
+                vector[j].u.cmpobj = NULL;
+                j++;
+            }
         } else {
-            ln = zsl->header->level[0].forward;
-            if (start > 0)
-                ln = zslGetElementByRank(zsl,start+1);
-        }
+            zset *zs = sortval->ptr;
+            zskiplist *zsl = zs->zsl;
+            zskiplistNode *ln;
 
-        while(rangelen--) {
-            serverAssertWithInfo(c,sortval,ln != NULL);
-            sdsele = zslGetNodeElement(ln);
-            vector[j].obj = createStringObject(sdsele,sdslen(sdsele));
-            vector[j].u.score = 0;
-            vector[j].u.cmpobj = NULL;
-            j++;
-            ln = desc ? ln->backward : ln->level[0].forward;
+            if (desc) {
+                ln = zsl->tail;
+                if (start > 0)
+                    ln = zslGetElementByRank(zsl,zsetLength(sortval)-start);
+            } else {
+                ln = zsl->header->level[0].forward;
+                if (start > 0)
+                    ln = zslGetElementByRank(zsl,start+1);
+            }
+
+            while(rangelen--) {
+                serverAssertWithInfo(c,sortval,ln != NULL);
+                sds ele = zslGetNodeElement(ln);
+                vector[j].obj = createStringObject(ele,sdslen(ele));
+                vector[j].u.score = 0;
+                vector[j].u.cmpobj = NULL;
+                j++;
+                ln = desc ? ln->backward : ln->level[0].forward;
+            }
         }
         /* Fix start/end: output code is not aware of this optimization. */
         end -= start;
         start = 0;
     } else if (sortval->type == OBJ_ZSET) {
-        dict *set = ((zset*)sortval->ptr)->dict;
-        dictIterator di;
-        dictEntry *setele;
-        sds sdsele;
-
         if (server.memory_tracking_enabled)
             oldsize = kvobjAllocSize(sortval);
-        dictInitIterator(&di, set);
-        while((setele = dictNext(&di)) != NULL) {
-            sdsele = zslGetNodeElement(dictGetKey(setele));
-            vector[j].obj = createStringObject(sdsele,sdslen(sdsele));
-            vector[j].u.score = 0;
-            vector[j].u.cmpobj = NULL;
-            j++;
+        if (sortval->encoding == OBJ_ENCODING_BTREE) {
+            zbtreeIterator iter;
+            const unsigned char *ele;
+            size_t len;
+            zbtreeIteratorStart(sortval->ptr, 0, &iter);
+            while(zbtreeIteratorNext(&iter, 0, &ele, &len, NULL)) {
+                vector[j].obj = createStringObject((char *)ele,len);
+                vector[j].u.score = 0;
+                vector[j].u.cmpobj = NULL;
+                j++;
+            }
+        } else {
+            dict *set = ((zset*)sortval->ptr)->dict;
+            dictIterator di;
+            dictEntry *setele;
+            dictInitIterator(&di, set);
+            while((setele = dictNext(&di)) != NULL) {
+                sds ele = zslGetNodeElement(dictGetKey(setele));
+                vector[j].obj = createStringObject(ele,sdslen(ele));
+                vector[j].u.score = 0;
+                vector[j].u.cmpobj = NULL;
+                j++;
+            }
+            dictResetIterator(&di);
         }
-        dictResetIterator(&di);
         if (server.memory_tracking_enabled)
             updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), sortval, oldsize, kvobjAllocSize(sortval));
     } else {

src/valgrind.sup

@@ -24,3 +24,15 @@
    fun:ztrymalloc_usable
    fun:ztrymalloc
 }
+
+# The only pointer to a long zset member stored outside a B+ tree leaf lives
+# unaligned inside a packed leaf record, where the leak checker does not look.
+# Every such member is allocated by zbtExternalDup(), so match only that frame.
+{
+   <zset_btree_external_member>
+   Memcheck:Leak
+   match-leak-kinds: definite
+   ...
+   fun:zbtAlloc
+   fun:zbtExternalDup
+}

src/zset_btree.h

@@ -0,0 +1,141 @@
+/* zset_btree.h -- memory efficient sorted set implementation.
+ *
+ * Copyright (c) 2026-Present, Redis Ltd.
+ * All rights reserved.
+ *
+ * Licensed under your choice of (a) the Redis Source Available License 2.0
+ * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
+ * GNU Affero General Public License v3 (AGPLv3).
+ */
+
+#ifndef REDIS_ZSET_BTREE_H
+#define REDIS_ZSET_BTREE_H
+
+#include "sds.h"
+#include <stddef.h>
+#include <stdint.h>
+
+typedef struct zbtreeSet zbtreeSet;
+
+/* A ZADD lookup records both index positions. An insertion can reuse the hash
+ * position after a miss, while an update can reuse the score position after
+ * a hit. The contents are private to zset_btree.c. */
+typedef struct {
+    void *bucket;
+    unsigned int pos;
+    void *score_leaf;
+    unsigned int score_pos;
+    uint32_t hash;
+    uint32_t revision;
+} zbtreeInsertPosition;
+
+/* Iterators point inside score leaves. They and the member bytes returned by
+ * zbtreeIteratorNext() remain valid only until the set is modified. */
+typedef struct {
+    void *leaf;
+    unsigned int pos;
+} zbtreeIterator;
+
+/* This state exists only while active defragmentation is visiting a large
+ * sorted set. It is kept by the defragmenter, not in every zset. */
+typedef struct {
+    zbtreeSet *tree;
+    uint32_t score_leaf_id;
+    uint16_t score_pos;
+    uint8_t leaf_moved;
+} zbtreeDefragState;
+
+typedef void *zbtreeDefragAllocFunction(void *ptr);
+typedef void zbtreeDefragFreeFunction(void *ptr);
+typedef void zbtreeScanFunction(void *privdata, const unsigned char *ele,
+                                size_t len, double score);
+
+/* Creation, release, and memory accounting. */
+zbtreeSet *zbtreeCreate(void);
+zbtreeSet *zbtreeDup(const zbtreeSet *zs);
+void zbtreeFree(zbtreeSet *zs);
+unsigned long zbtreeLength(const zbtreeSet *zs);
+size_t zbtreeAllocSize(const zbtreeSet *zs);
+void zbtreeDismissMemory(zbtreeSet *zs);
+void zbtreeReserve(zbtreeSet *zs, unsigned long count);
+int zbtreeTryReserve(zbtreeSet *zs, uint64_t count);
+
+/* Active defragmentation. Start may move the set itself and therefore returns
+ * its current address. Step moves at most one leaf, one external member, and
+ * the inner pages that finish at that leaf. It returns one when complete. */
+zbtreeSet *zbtreeDefragStart(zbtreeSet *zs,
+                             zbtreeDefragAllocFunction *defrag_alloc);
+void zbtreeDefragStateReset(zbtreeDefragState *state);
+int zbtreeDefragStep(zbtreeSet *zs, zbtreeDefragState *state,
+                     zbtreeDefragAllocFunction *defrag_alloc,
+                     zbtreeDefragFreeFunction *defrag_free,
+                     unsigned long *scanned);
+
+/* Lookup and modification. InsertNew* require an absent member. UpdateScore
+ * requires the position returned by a successful FindForAdd. Delete returns
+ * zero when the member is absent. */
+int zbtreeScore(zbtreeSet *zs, sds ele, double *score);
+int zbtreeScoreRaw(zbtreeSet *zs, const unsigned char *ele,
+                   size_t elelen, double *score);
+/* Look up up to 16 scores. Bit i of the result says whether scores[i] exists. */
+#define ZBT_SCORE_BATCH_SIZE 16
+unsigned int zbtreeScores(zbtreeSet *zs, const sds *members, unsigned int count,
+                          double *scores);
+int zbtreeFindForAdd(zbtreeSet *zs, sds ele, double *score,
+                     zbtreeInsertPosition *position);
+long zbtreeRank(zbtreeSet *zs, sds ele, int reverse, double *score);
+void zbtreeInsertNew(zbtreeSet *zs, double score, sds ele,
+                     const zbtreeInsertPosition *position);
+void zbtreeInsertNewRaw(zbtreeSet *zs, double score, const unsigned char *ele,
+                        size_t elelen, const zbtreeInsertPosition *position);
+void zbtreeInsertNewAppend(zbtreeSet *zs, double score,
+                           const unsigned char *ele, size_t elelen);
+void zbtreeInsertNewAppendWithHash(zbtreeSet *zs, double score,
+                                   const unsigned char *ele, size_t elelen,
+                                   uint32_t hash);
+/* Buffers up to one leaf's worth of zbtreeInsertNewAppend()-style inserts and
+ * flushes them with a single bulk leaf build instead of one insert per
+ * element -- see the comment above its implementation in zset_btree.c for
+ * why. Same precondition as zbtreeInsertNewAppend(): every added element
+ * must sort after everything already in the tree and be known absent. */
+typedef struct zbtreeAppendBatch zbtreeAppendBatch;
+zbtreeAppendBatch *zbtreeAppendBatchCreate(zbtreeSet *zs);
+void zbtreeAppendBatchAdd(zbtreeAppendBatch *b, double score,
+                          const unsigned char *ele, size_t elelen);
+void zbtreeAppendBatchAddWithHash(zbtreeAppendBatch *b, double score,
+                                  const unsigned char *ele, size_t elelen,
+                                  uint32_t hash);
+void zbtreeAppendBatchFinish(zbtreeAppendBatch *b);
+int zbtreeDelete(zbtreeSet *zs, sds ele);
+void zbtreeDeleteEdge(zbtreeSet *zs, int reverse);
+void zbtreeUpdateScore(zbtreeSet *zs, sds ele, double score,
+                       const zbtreeInsertPosition *position);
+unsigned long zbtreeDeleteRangeByRank(zbtreeSet *zs, unsigned long start,
+                                      unsigned long end);
+unsigned long zbtreeDeleteRangeByScore(zbtreeSet *zs, double min, int minex,
+                                       double max, int maxex);
+
+/* Ordered access. A successful seek leaves the iterator on its result; Next
+ * returns that result and advances in the requested direction. Its member,
+ * length, and score output pointers may be NULL. */
+int zbtreeIteratorSeekRank(const zbtreeSet *zs, unsigned long rank,
+                           zbtreeIterator *iter);
+int zbtreeIteratorSeekScore(const zbtreeSet *zs, double score, int exclusive,
+                            int reverse, zbtreeIterator *iter,
+                            unsigned long *rank);
+int zbtreeIteratorSeekLex(const zbtreeSet *zs, sds ele, int exclusive,
+                          int reverse, zbtreeIterator *iter,
+                          unsigned long *rank);
+int zbtreeIteratorStart(const zbtreeSet *zs, int reverse,
+                        zbtreeIterator *iter);
+int zbtreeIteratorNext(zbtreeIterator *iter, int reverse,
+                       const unsigned char **ele, size_t *len, double *score);
+unsigned long zbtreeCountByScore(const zbtreeSet *zs, double min, int minex,
+                                 double max, int maxex);
+
+/* Incremental unordered access. The cursor is an opaque hash position. */
+uint64_t zbtreeScan(zbtreeSet *zs, uint64_t cursor,
+                    unsigned long count, zbtreeScanFunction *fn,
+                    void *privdata);
+
+#endif

tests/integration/convert-ziplist-zset-on-load.tcl

@@ -16,10 +16,10 @@ start_server [list overrides [list "dir" $server_path "dbfilename" "zset-ziplist
 
 exec cp -f tests/assets/zset-ziplist.rdb $server_path
 start_server [list overrides [list "dir" $server_path "dbfilename" "zset-ziplist.rdb" "zset-max-ziplist-entries" 1]] {
-    test "RDB load ziplist zset: converts to skiplist when zset-max-ziplist-entries is exceeded" {
+    test "RDB load ziplist zset: converts to btree when zset-max-ziplist-entries is exceeded" {
         r select 0
 
-        assert_encoding skiplist zset
+        assert_encoding btree zset
         assert_equal 2 [r zcard zset]
         assert_match {one 1 two 2} [r zrange zset 0 -1 withscores]
     }

tests/integration/corrupt-dump.tcl

@@ -18,6 +18,63 @@ set run_oom_tests [expr {($arch_name == "x86_64" || $arch_name == "aarch64") &&
 
 set corrupt_payload_7445 "\x0E\x01\x1D\x1D\x00\x00\x00\x16\x00\x00\x00\x03\x00\x00\x04\x43\x43\x43\x43\x06\x04\x42\x42\x42\x42\x06\x3F\x41\x41\x41\x41\xFF\x09\x00\x88\xA5\xCA\xA8\xC5\x41\xF4\x35"
 
+test {corrupt payload: zset reservation rejects oversized counts} {
+    start_server {overrides {loglevel verbose}} {
+        r debug set-skip-checksum-validation 1
+        r set target preserved
+        # Both legacy ZSET and ZSET_2 use the same 64-bit count. These values
+        # exercise sizing overflow and allocator rejection without attempting
+        # a large allocation. The footer has RDB version 9 and a dummy CRC.
+        foreach type {03 05} {
+            foreach count {fffffffffffffffe 1fffffffffffffff 1000000000000000} {
+                set payload [binary format H* "${type}81${count}09000000000000000000"]
+                assert_error {*Bad data format*} {r restore target 0 $payload replace}
+                assert_equal preserved [r get target]
+                assert_equal PONG [r ping]
+            }
+        }
+        verify_log_message 0 "*Unable to reserve zset index*" 0
+    }
+}
+
+test {Zset listpack scores are validated before loading or conversion} {
+    start_server {} {
+        r zadd score-source 1.5 a 2.5 b
+        assert_encoding listpack score-source
+        set payload [r dump score-source]
+        # Mutate only the score text; bypass the now-stale DUMP checksum so
+        # the test exercises semantic validation of a well-formed listpack.
+        r debug set-skip-checksum-validation 1
+        # Without sanitization a listpack that stays one is trusted, as for
+        # other types. A conversion still checks every score first.
+        foreach {sanitize limit} {yes 128 yes 1 no 1} {
+            r config set sanitize-dump-payload $sanitize
+            r config set zset-max-listpack-entries $limit
+            foreach score {nan NaN abc 1.x} {
+                set invalid [string map [list 1.5 $score] $payload]
+                assert_error {*Bad data format*} {r restore invalid-score 0 $invalid}
+                assert_equal 0 [r exists invalid-score]
+                assert_equal PONG [r ping]
+            }
+            # Both infinities are legal and must remain reloadable.
+            foreach score {inf -inf} {
+                # -inf needs four bytes, so use ZADD's genuine payload.
+                r del score-valid
+                r zadd score-valid $score a 2.5 b
+                set valid [r dump score-valid]
+                r restore score-roundtrip 0 $valid replace
+                assert_equal $score [r zscore score-roundtrip a]
+            }
+        }
+        r debug set-skip-checksum-validation 0
+        r config set sanitize-dump-payload yes
+        r config set zset-max-listpack-entries 128
+        r debug reload
+        assert_equal PONG [r ping]
+        assert_equal 0 [r exists invalid-score]
+    }
+}
+
 test {corrupt payload: #7445 - with sanitize} {
     start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] {
         r config set sanitize-dump-payload yes
@@ -1214,4 +1271,3 @@ test {corrupt payload: stream XTRIM with non-minimally encoded flags} {
 }
 
 } ;# tags
-

tests/integration/dismiss-mem.tcl

@@ -32,10 +32,16 @@ start_server {tags {"dismiss external:skip needs:debug"}} {
         }
 
         # zset
-        r zadd bigzset1 1.0 $bigstr         ; # skiplist encoding
+        r zadd bigzset1 1.0 $bigstr         ; # B+ tree, external member
+        assert_encoding btree bigzset1
         for {set i 0} {$i < 128} {incr i} {
-            r zadd bigzset2 1.0 $64bytes    ; # ziplist encoding
+            r zadd bigzset2 1.0 [format "%064d" $i]
         }
+        assert_encoding listpack bigzset2
+        for {set i 0} {$i < 512} {incr i} {
+            r zadd bigzset3 $i "${64bytes}:$i"
+        }
+        assert_encoding btree bigzset3
 
         # hash
         r hset bighash1 field1 $bigstr      ; # hash encoding

tests/modules/scan.c

@@ -103,6 +103,52 @@ int scan_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     return REDISMODULE_OK;
 }
 
+typedef struct {
+    RedisModuleCtx *ctx;
+    RedisModuleString *first;
+    size_t count;
+} scan_key_delete_pd;
+
+void scan_key_delete_callback(RedisModuleKey *key, RedisModuleString *field,
+                              RedisModuleString *value, void *privdata)
+{
+    REDISMODULE_NOT_USED(value);
+    scan_key_delete_pd *pd = privdata;
+
+    RedisModule_ReplyWithString(pd->ctx, field);
+    pd->count++;
+    if (pd->count == 1) {
+        pd->first = field;
+        RedisModule_RetainString(pd->ctx, field);
+    } else if (pd->count == 2) {
+        int deleted;
+        RedisModule_ZsetRem(key, pd->first, &deleted);
+        RedisModule_FreeString(pd->ctx, pd->first);
+        pd->first = NULL;
+    }
+}
+
+/* Scan a sorted set while deleting the first member from the second callback.
+ * This exercises the documented guarantee that deleting a returned element
+ * does not hide elements that remain present throughout the scan. */
+int scan_key_delete(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
+{
+    if (argc != 2) return RedisModule_WrongArity(ctx);
+
+    RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ | REDISMODULE_WRITE);
+    if (!key) return RedisModule_ReplyWithError(ctx, "not found");
+
+    scan_key_delete_pd pd = {.ctx = ctx};
+    RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
+    RedisModuleScanCursor *cursor = RedisModule_ScanCursorCreate();
+    while (RedisModule_ScanKey(key, cursor, scan_key_delete_callback, &pd));
+    RedisModule_ScanCursorDestroy(cursor);
+    if (pd.first) RedisModule_FreeString(ctx, pd.first);
+    RedisModule_ReplySetArrayLength(ctx, pd.count);
+    RedisModule_CloseKey(key);
+    return REDISMODULE_OK;
+}
+
 int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     REDISMODULE_NOT_USED(argv);
     REDISMODULE_NOT_USED(argc);
@@ -115,7 +161,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     if (RedisModule_CreateCommand(ctx, "scan.scan_key", scan_key, "", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx, "scan.scan_key_delete", scan_key_delete, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     return REDISMODULE_OK;
 }
-
-

tests/modules/zset.c

@@ -1,6 +1,7 @@
 #include "redismodule.h"
 #include <math.h>
 #include <errno.h>
+#include <strings.h>
 
 /* ZSET.REM key element
  *
@@ -69,6 +70,112 @@ int zset_incrby(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
         return RedisModule_ReplyWithError(ctx, "ERR ZsetIncrby failed");
 }
 
+/* Range APIs take doubles directly, so expose NaN bounds even though
+ * RedisModule_StringToDouble() rejects them. */
+static int zset_range_bound(RedisModuleString *arg, double *value) {
+    size_t len;
+    const char *str = RedisModule_StringPtrLen(arg, &len);
+    if (len == 3 && !strncasecmp(str, "nan", 3)) {
+        *value = NAN;
+        return REDISMODULE_OK;
+    }
+    if (len == 4 && !strncasecmp(str, "-nan", 4)) {
+        *value = -NAN;
+        return REDISMODULE_OK;
+    }
+    return RedisModule_StringToDouble(arg, value);
+}
+
+/* ZSET.RANGE key score first|last min max minex maxex
+ * ZSET.RANGE key lex   first|last min max
+ *
+ * Exposes the module range iterator to the test suite. The reply contains
+ * member, score pairs in the direction selected by first or last. */
+int zset_range(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 6 && argc != 8) return RedisModule_WrongArity(ctx);
+    RedisModule_AutoMemory(ctx);
+
+    size_t mode_len, direction_len;
+    const char *mode = RedisModule_StringPtrLen(argv[2], &mode_len);
+    const char *direction = RedisModule_StringPtrLen(argv[3], &direction_len);
+    int reverse;
+    if (direction_len == 5 && !strncasecmp(direction, "first", 5))
+        reverse = 0;
+    else if (direction_len == 4 && !strncasecmp(direction, "last", 4))
+        reverse = 1;
+    else
+        return RedisModule_ReplyWithError(ctx, "ERR invalid direction");
+
+    RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
+    int status;
+    if (mode_len == 5 && !strncasecmp(mode, "score", 5) && argc == 8) {
+        double min, max;
+        long long minex, maxex;
+        if (zset_range_bound(argv[4], &min) == REDISMODULE_ERR ||
+            zset_range_bound(argv[5], &max) == REDISMODULE_ERR ||
+            RedisModule_StringToLongLong(argv[6], &minex) == REDISMODULE_ERR ||
+            RedisModule_StringToLongLong(argv[7], &maxex) == REDISMODULE_ERR ||
+            (minex != 0 && minex != 1) || (maxex != 0 && maxex != 1))
+            return RedisModule_ReplyWithError(ctx, "ERR invalid score range");
+        status = reverse ?
+            RedisModule_ZsetLastInScoreRange(key, min, max, minex, maxex) :
+            RedisModule_ZsetFirstInScoreRange(key, min, max, minex, maxex);
+    } else if (mode_len == 3 && !strncasecmp(mode, "lex", 3) && argc == 6) {
+        status = reverse ?
+            RedisModule_ZsetLastInLexRange(key, argv[4], argv[5]) :
+            RedisModule_ZsetFirstInLexRange(key, argv[4], argv[5]);
+    } else {
+        return RedisModule_ReplyWithError(ctx, "ERR invalid range mode");
+    }
+    if (status == REDISMODULE_ERR)
+        return RedisModule_ReplyWithError(ctx, "ERR range setup failed");
+
+    RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
+    long count = 0;
+    while (!RedisModule_ZsetRangeEndReached(key)) {
+        double score;
+        RedisModuleString *member = RedisModule_ZsetRangeCurrentElement(key, &score);
+        if (member == NULL) break;
+        RedisModule_ReplyWithString(ctx, member);
+        RedisModule_ReplyWithDouble(ctx, score);
+        RedisModule_FreeString(ctx, member);
+        count += 2;
+        if (reverse)
+            RedisModule_ZsetRangePrev(key);
+        else
+            RedisModule_ZsetRangeNext(key);
+    }
+    RedisModule_ZsetRangeStop(key);
+    RedisModule_ReplySetArrayLength(ctx, count);
+    return REDISMODULE_OK;
+}
+
+/* Invalidate the rank saved by a B+ tree range iterator. The iterator API
+ * should report the end of the range rather than aborting Redis. */
+int zset_range_mutate(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) return RedisModule_WrongArity(ctx);
+    RedisModule_AutoMemory(ctx);
+
+    RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ | REDISMODULE_WRITE);
+    if (RedisModule_ZsetLastInScoreRange(key, REDISMODULE_NEGATIVE_INFINITE,
+                                         REDISMODULE_POSITIVE_INFINITE, 0, 0) == REDISMODULE_ERR)
+        return RedisModule_ReplyWithError(ctx, "ERR range setup failed");
+
+    /* Keep one member so the open key handle itself remains valid. */
+    RedisModuleCallReply *reply = RedisModule_Call(ctx, "ZREMRANGEBYRANK", "sll",
+                                                   argv[1], (long long)0, (long long)-2);
+    if (reply == NULL || RedisModule_CallReplyType(reply) == REDISMODULE_REPLY_ERROR)
+        return RedisModule_ReplyWithError(ctx, "ERR range mutation failed");
+
+    RedisModuleString *member = RedisModule_ZsetRangeCurrentElement(key, NULL);
+    if (member != NULL || RedisModule_ZsetRangePrev(key) != 0 ||
+        !RedisModule_ZsetRangeEndReached(key))
+        return RedisModule_ReplyWithError(ctx, "ERR invalid iterator state");
+
+    RedisModule_ZsetRangeStop(key);
+    return RedisModule_ReplyWithSimpleString(ctx, "OK");
+}
+
 /* Structure to hold data for the delall scan callback */
 typedef struct {
     RedisModuleCtx *ctx;
@@ -172,6 +279,14 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
                                   1, 1, 1) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx, "zset.range", zset_range, "readonly",
+                                  1, 1, 1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "zset.range-mutate", zset_range_mutate, "write",
+                                  1, 1, 1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     if (RedisModule_CreateCommand(ctx, "zset.delall", zset_delall, "write touches-arbitrary-keys",
                                   0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;

tests/unit/aofrw.tcl

@@ -166,7 +166,7 @@ start_server {tags {"aofrw external:skip debug_defrag:skip"} overrides {aof-use-
     }
 
     foreach d {string int} {
-        foreach e {listpack skiplist} {
+        foreach e {listpack btree} {
             test "AOF rewrite of zset with $e encoding, $d data" {
                 r flushall
                 if {$e eq {listpack}} {set len 10} else {set len 1000}
@@ -191,6 +191,28 @@ start_server {tags {"aofrw external:skip debug_defrag:skip"} overrides {aof-use-
         }
     }
 
+    test {AOF rewrite of zset with external members} {
+        r flushall
+        set members {}
+        for {set j 0} {$j < 600} {incr j} {
+            set len [lindex {319 320 321 4096} [expr {$j % 4}]]
+            set prefix [format "%04d:" $j]
+            set member "$prefix[string repeat x [expr {$len-[string length $prefix]}]]"
+            lappend members $member
+            r zadd key $j $member
+        }
+        assert_encoding btree key
+        set expected [r zrange key 0 -1 withscores]
+
+        r bgrewriteaof
+        waitForBgrewriteaof r
+        r debug loadaof
+
+        assert_encoding btree key
+        assert_equal $expected [r zrange key 0 -1 withscores]
+        assert_equal 599 [r zscore key [lindex $members 599]]
+    }
+
     test "AOF rewrite functions" {
         r flushall
         r FUNCTION LOAD {#!lua name=test

tests/unit/keyspace.tcl

@@ -307,14 +307,14 @@ foreach {type large} [array get largevalue] {
         assert_equal $digest [debug_digest_value newzset1{t}]
     }
 
-     test {COPY basic usage for skiplist sorted set} {
+     test {COPY basic usage for btree sorted set} {
         r del zset2{t} newzset2{t}
         set original_max [lindex [r config get zset-max-ziplist-entries] 1]
         r config set zset-max-ziplist-entries 0
         for {set j 0} {$j < 130} {incr j} {
             r zadd zset2{t} [randomInt 50] ele-[randomInt 10]
         }
-        assert_encoding skiplist zset2{t}
+        assert_encoding btree zset2{t}
         r copy zset2{t} newzset2{t}
         set digest [debug_digest_value zset2{t}]
         assert_equal $digest [debug_digest_value newzset2{t}]
@@ -325,6 +325,35 @@ foreach {type large} [array get largevalue] {
         r config set zset-max-ziplist-entries $original_max
     }
 
+    test {COPY duplicates all btree pages while its member table grows} {
+        r del zset-copy-source{t} zset-copy-dest{t}
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+
+        # The last insertion starts a member-table resize and leaves many
+        # score pages still to copy into the new table.
+        for {set j 0} {$j < 3969} {incr j} {
+            r zadd zset-copy-source{t} [expr {$j * 17 % 5003}] \
+                [format "member:%05d:abcdefghijklmnop" $j]
+        }
+        assert_encoding btree zset-copy-source{t}
+        r copy zset-copy-source{t} zset-copy-dest{t}
+        set digest [debug_digest_value zset-copy-source{t}]
+        assert_equal $digest [debug_digest_value zset-copy-dest{t}]
+
+        # Free the source before exercising every copied lookup and rank.
+        r del zset-copy-source{t}
+        assert_equal $digest [debug_digest_value zset-copy-dest{t}]
+        for {set j 0} {$j < 3969} {incr j 31} {
+            set member [format "member:%05d:abcdefghijklmnop" $j]
+            assert {[r zscore zset-copy-dest{t} $member] ne {}}
+            assert {[r zrank zset-copy-dest{t} $member] ne {}}
+            r zincrby zset-copy-dest{t} 0.5 $member
+        }
+        assert_equal 3969 [r zcard zset-copy-dest{t}]
+        r config set zset-max-listpack-entries $original_max
+    }
+
     test {COPY basic usage for listpack hash} {
         r config set hash-max-listpack-entries 512
         r del hash1{t} newhash1{t}

tests/unit/moduleapi/scan.tcl

@@ -95,13 +95,28 @@ start_server {tags {"modules external:skip"}} {
         lsort [r scan.scan_key zz]
     } {{f1 1} {f2 2}}
 
-    test {Module scan zset skiplist} {
+    test {Module scan zset btree} {
         r config set zset-max-ziplist-entries 2
         r zadd zz 3 f3
-        assert_encoding skiplist zz
+        assert_encoding btree zz
         lsort [r scan.scan_key zz]
     } {{f1 1} {f2 2} {f3 3}}
 
+    test {Module scan zset btree tolerates deleting a returned member} {
+        r del zz
+        for {set i 0} {$i < 100} {incr i} {
+            r zadd zz $i [format "member:%03d" $i]
+        }
+        assert_encoding btree zz
+
+        set visited [lsort -unique [r scan.scan_key_delete zz]]
+        set remaining [r zrange zz 0 -1]
+        assert_equal 99 [llength $remaining]
+        foreach member $remaining {
+            assert {$member in $visited}
+        }
+    }
+
     test {Module scan set intset} {
         r sadd ss 1 2
         assert_encoding intset ss
@@ -124,4 +139,4 @@ start_server {tags {"modules external:skip"}} {
     test "Unload the module - scan" {
         assert_equal {OK} [r module unload scan]
     }
-}
+}

tests/unit/moduleapi/zset.tcl

@@ -34,6 +34,73 @@ start_server {tags {"modules external:skip"}} {
         assert_equal {hello 100} [r zrange k 0 -1 withscores]
     }
 
+    test {Module zset range iteration with btree encoding} {
+        set old_max_entries [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+
+        r del k
+        r zadd k 1 a 2 b 2 c 3 d 4 e
+        assert_encoding btree k
+        assert_equal {d 3 e 4} [r zset.range k score first 2 4 1 0]
+        assert_equal {e 4 d 3} [r zset.range k score last 2 4 1 0]
+        assert_equal {} [r zset.range k score first 10 20 0 0]
+
+        r del k
+        r zadd k 0 alpha 0 beta 0 delta 0 gamma
+        assert_encoding btree k
+        assert_equal {beta 0 delta 0} \
+            [r zset.range k lex first \[beta \(gamma]
+        assert_equal {delta 0 beta 0} \
+            [r zset.range k lex last \[beta \(gamma]
+        assert_equal {} [r zset.range k lex first \[z +]
+
+        r config set zset-max-listpack-entries $old_max_entries
+    }
+
+    foreach encoding {listpack btree} {
+        foreach count {3 1000} {
+            test "Module zset NaN score bounds are empty - $encoding, $count members" {
+                set old_max_entries [lindex [r config get zset-max-listpack-entries] 1]
+                r config set zset-max-listpack-entries [expr {$encoding eq "listpack" ? 2000 : 0}]
+                r del k
+                set elements {}
+                for {set j 1} {$j <= $count} {incr j} {
+                    lappend elements $j member:$j
+                }
+                r zadd k {*}$elements
+                assert_encoding $encoding k
+
+                foreach direction {first last} {
+                    foreach {min max} {NaN 10 0 NaN NaN NaN -NaN 10 0 -NaN -NaN -NaN} {
+                        foreach minex {0 1} {
+                            foreach maxex {0 1} {
+                                assert_equal {} [r zset.range k score $direction $min $max $minex $maxex]
+                            }
+                        }
+                    }
+                }
+                assert_equal {member:1 1 member:2 2 member:3 3} \
+                    [r zset.range k score first -inf 3 0 0]
+                assert_equal {member:3 3 member:2 2 member:1 1} \
+                    [r zset.range k score last -inf 3 0 0]
+                r config set zset-max-listpack-entries $old_max_entries
+            }
+        }
+    }
+
+    test {Module zset range iteration ends after its saved rank is invalidated} {
+        set old_max_entries [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+
+        r del k
+        r zadd k 1 a 2 b 3 c
+        assert_encoding btree k
+        assert_equal OK [r zset.range-mutate k]
+        assert_equal {c} [r zrange k 0 -1]
+
+        r config set zset-max-listpack-entries $old_max_entries
+    }
+
     test {Module zset - KEYSIZES is updated as expected (like test at hash.tcl)} {
         proc run_cmd_verify_hist {cmd expOutput {retries 1}} {
             proc K {} {return [string map { "db0_distrib_zsets_items" "db0_ZSET" "# Keysizes" "" " " "" "\n" "" "\r" "" } [r info keysizes] ]}

tests/unit/scan.tcl

@@ -308,7 +308,7 @@ proc test_scan {type} {
         }
     }
 
-    foreach enc {listpack skiplist} {
+    foreach enc {listpack btree} {
         test "{$type} ZSCAN with encoding $enc" {
             # Create the Sorted Set
             r del zset
@@ -348,6 +348,293 @@ proc test_scan {type} {
         }
     }
 
+    if {$type eq {standalone}} {
+        test {B+ tree ZSCAN small and streaming replies preserve framing} {
+            foreach proto {2 3} {
+                r hello $proto
+                foreach size {1 31 32 33 129} {
+                    r del zset
+                    for {set j 0} {$j < 129} {incr j} {
+                        r zadd zset 1.25 member:$j
+                    }
+                    for {set j $size} {$j < 129} {incr j} {
+                        r zrem zset member:$j
+                    }
+                    assert_encoding btree zset
+                    r readraw 1
+                    assert_equal {*2} [r zscan zset 0 count 10000]
+                    assert_equal {$1} [r read]
+                    assert_equal 0 [r read]
+                    set header [r read]
+                    assert_equal {*} [string index $header 0]
+                    set fields [string range $header 1 end]
+                    assert {$fields % 2 == 0}
+                    set seen {}
+                    for {set j 0} {$j < $fields / 2} {incr j} {
+                        assert_match {$*} [r read]
+                        dict set seen [r read] 1
+                        assert_equal {$4} [r read]
+                        assert_equal 1.25 [r read]
+                    }
+                    assert_equal $size [dict size $seen]
+                    r readraw 0
+                    assert_equal {0 {}} [r zscan zset 0 count 10000 match absent:*]
+                    assert_equal PONG [r ping]
+                    set rd [redis_deferring_client]
+                    $rd hello $proto
+                    $rd read
+                    $rd client reply off
+                    $rd zscan zset 0 count 10000
+                    $rd client reply on
+                    assert_equal OK [$rd read]
+                    $rd client reply skip
+                    $rd zscan zset 0 count 10000
+                    $rd ping
+                    assert_equal PONG [$rd read]
+                    $rd close
+                }
+            }
+            r hello 2
+            set _ {}
+        } {} {resp3}
+
+        test {B+ tree ZSCAN headers preserve queued reply order} {
+            r del zset{t}
+            for {set j 0} {$j < 129} {incr j} {
+                r zadd zset{t} 1.25 member:$j
+            }
+            assert_encoding btree zset{t}
+            foreach proto {2 3} {
+                r hello $proto
+                foreach size {0 1024 65536} {
+                    set preceding [string repeat x $size]
+                    r set preceding{t} $preceding
+                    r multi
+                    r getrange preceding{t} 0 -1
+                    r zscan zset{t} 0 count 10000 match absent:*
+                    r zscan zset{t} 0 count 10000
+                    r ping
+                    set replies [r exec]
+                    assert_equal $preceding [lindex $replies 0]
+                    assert_equal {0 {}} [lindex $replies 1]
+                    assert_equal 0 [lindex $replies 2 0]
+                    set seen {}
+                    foreach {member score} [lindex $replies 2 1] {
+                        assert_equal 1.25 $score
+                        dict set seen $member 1
+                    }
+                    assert_equal 129 [dict size $seen]
+                    assert_equal PONG [lindex $replies 3]
+                }
+            }
+            r hello 2
+            set _ {}
+        } {} {resp3}
+
+        test {B+ tree ZSCAN keeps scores as bulk strings in RESP3} {
+            r del zset
+            set elements {}
+            for {set j 0} {$j < 129} {incr j} {
+                set score [expr {$j == 0 ? 3.3 : 1.25}]
+                lappend elements $score member:$j
+            }
+            r zadd zset {*}$elements
+            assert_encoding btree zset
+
+            r hello 3
+            r readraw 1
+            assert_equal {*2} [r zscan zset 0 count 10000]
+            assert_equal {$1} [r read]
+            assert_equal 0 [r read]
+            set array_header [r read]
+            assert_equal {*} [string index $array_header 0]
+            set fields [string range $array_header 1 end]
+            assert {$fields % 2 == 0}
+            set seen {}
+            for {set j 0} {$j < $fields / 2} {incr j} {
+                assert_match {$*} [r read]
+                set member [r read]
+                set score_header [r read]
+                set score [r read]
+                dict set seen $member 1
+                if {$member eq {member:0}} {
+                    assert_equal {$3} $score_header
+                    assert_equal 3.3 $score
+                } else {
+                    assert_equal {$4} $score_header
+                    assert_equal 1.25 $score
+                }
+            }
+            assert_equal 129 [dict size $seen]
+            r readraw 0
+            r hello 2
+            set _ {}
+        } {} {resp3}
+    }
+
+    test "{$type} ZSCAN does not skip members after deleting returned members" {
+        r del zset
+        set elements {}
+        for {set j 0} {$j < 1000} {incr j} {
+            lappend elements $j key:$j
+        }
+        r zadd zset {*}$elements
+        assert_encoding btree zset
+
+        lassign [r zscan zset 0 count 10] cursor items
+        set seen {}
+        foreach {member score} $items {
+            lappend seen $member
+            r zrem zset $member
+        }
+        while {$cursor != 0} {
+            lassign [r zscan zset $cursor count 10] cursor items
+            foreach {member score} $items {
+                lappend seen $member
+            }
+        }
+
+        assert_equal 1000 [llength [lsort -unique $seen]]
+    }
+
+    test "{$type} ZSCAN does not skip members changed between calls" {
+        r del zset
+        set elements {}
+        set members {}
+        for {set j 0} {$j < 2048} {incr j} {
+            set member [format "stable:%05d" $j]
+            lappend elements $j $member
+            lappend members $member
+        }
+        r zadd zset {*}$elements
+        assert_encoding btree zset
+
+        set cursor 0
+        set seen {}
+        set calls 0
+        while 1 {
+            lassign [r zscan zset $cursor count 1] cursor items
+            foreach {member score} $items {
+                lappend seen $member
+            }
+            if {$cursor == 0} break
+            incr calls
+
+            # Moving a member can change its score leaf, but it remains part of
+            # the set for the complete scan and must eventually be returned.
+            set member [lindex $members [expr {($calls * 31) % 2048}]]
+            r zadd zset [expr {2048 + $calls}] $member
+        }
+
+        assert_equal 2048 [llength [lsort -unique $seen]]
+    }
+
+    test "{$type} ZSCAN does not skip members while its table grows" {
+        r del zset
+        set elements {}
+        for {set j 0} {$j < 1984} {incr j} {
+            lappend elements $j [format "stable:%05d" $j]
+        }
+        r zadd zset {*}$elements
+        assert_encoding btree zset
+
+        set cursor 0
+        set seen {}
+        set calls 0
+        while 1 {
+            lassign [r zscan zset $cursor count 1] cursor items
+            foreach {member score} $items {
+                if {[string match "stable:*" $member]} {
+                    lappend seen $member
+                }
+            }
+            if {$cursor == 0} break
+
+            # A 256-bucket narrow table has 2048 slots and grows when the
+            # 1985th member is added. Further additions advance the copy.
+            incr calls
+            r zadd zset [expr {1984 + $calls}] [format "added:%05d" $calls]
+        }
+
+        assert_equal 1984 [llength [lsort -unique $seen]]
+    }
+
+    test "{$type} ZSCAN finishes while a bounded set keeps replacing its tables" {
+        # Between calls the set grows and is cut back by a large range
+        # deletion, which installs new member tables every time. A cursor
+        # tied to one table restarted forever. This one must finish and
+        # return every member that stays in the set.
+        r del zset
+        set elements {}
+        for {set j 0} {$j < 200} {incr j} {
+            lappend elements $j [format "anchor:%03d" $j]
+        }
+        r zadd zset {*}$elements
+        assert_encoding btree zset
+        set temps {}
+        for {set j 0} {$j < 3000} {incr j} {
+            lappend temps [expr {1000000 + $j}] temp:$j
+        }
+
+        set cursor 0
+        set seen {}
+        set calls 0
+        while 1 {
+            lassign [r zscan zset $cursor count 10] cursor items
+            foreach {member score} $items {
+                if {[string match "anchor:*" $member]} {
+                    lappend seen $member
+                }
+            }
+            # Clients that keep cursors as doubles need them below 2^53.
+            assert {$cursor < 9007199254740992}
+            if {$cursor == 0} break
+            incr calls
+            assert {$calls < 1000}
+            r zadd zset {*}$temps
+            r zremrangebyscore zset 1000000 +inf
+        }
+        assert_equal 200 [llength [lsort -unique $seen]]
+    }
+
+    test "{$type} ZSCAN finishes under constant size churn" {
+        # Replacing members leaves tombstones, and their cleanup installs a
+        # same-size table every few hundred calls, before a complete scan.
+        r del zset
+        set elements {}
+        for {set j 0} {$j < 5000} {incr j} {
+            lappend elements $j [format "anchor:%05d" $j]
+            lappend elements $j churn:$j
+        }
+        r zadd zset {*}$elements
+        assert_encoding btree zset
+
+        set cursor 0
+        set seen {}
+        set calls 0
+        set next 5000
+        while 1 {
+            lassign [r zscan zset $cursor count 10] cursor items
+            foreach {member score} $items {
+                if {[string match "anchor:*" $member]} {
+                    lappend seen $member
+                }
+            }
+            if {$cursor == 0} break
+            incr calls
+            assert {$calls < 2000}
+            set add {}
+            set rem {}
+            for {set j 0} {$j < 200} {incr j; incr next} {
+                lappend add $next churn:$next
+                lappend rem churn:[expr {$next - 5000}]
+            }
+            r zadd zset {*}$add
+            r zrem zset {*}$rem
+        }
+        assert_equal 5000 [llength [lsort -unique $seen]]
+    }
+
     test "{$type} SCAN guarantees check under write load" {
         r flushdb
         populate 100
@@ -425,7 +712,7 @@ proc test_scan {type} {
         assert {$first_score != 0}
     }
 
-    foreach enc {listpack skiplist} {
+    foreach enc {listpack btree} {
         test "{$type} ZSCAN scores match ZSCORE with encoding $enc" {
             r del mykey
             # 9.8813129168249309e-323 is the denormal from the #2175 test above;
@@ -435,7 +722,7 @@ proc test_scan {type} {
             r zadd mykey 3.3 a 1 b 1.25 c -0.1 d 1e100 e 9.8813129168249309e-323 f \
                 inf g -inf h -1.2345678901234567e23 i 1152921504606846976 j \
                 123456789.12345679 k
-            if {$enc eq {skiplist}} {
+            if {$enc eq {btree}} {
                 # Push the set past zset-max-listpack-entries (default 128).
                 set elements {}
                 for {set j 0} {$j < 200} {incr j} {
@@ -482,9 +769,9 @@ proc test_scan {type} {
             lappend elements $j $member
         }
         r zadd mykey {*}$elements
-        assert_encoding skiplist mykey
+        assert_encoding btree mykey
 
-        # Exceed the stack vector's capacity and delete the nodes before
+        # Exercise a large streaming reply and delete the nodes before
         # EXEC flushes the reply. The output buffer must own the returned bytes.
         r multi
         r zscan mykey 0 COUNT 1000

tests/unit/type/zset.tcl

@@ -6,6 +6,44 @@ start_server {tags {"zset"}} {
         }
     }
 
+    test {Small ZSET_2 loads preserve signed zero when a long member needs a B+tree} {
+        set old_limit [lindex [r config get zset-max-listpack-entries] 1]
+        set longmember [string repeat L 65]
+        foreach count {2 128 129} {
+            foreach longscore {-1 1} {
+                r del zero-source{t} zero-restored{t}
+                r config set zset-max-listpack-entries 0
+                r zadd zero-source{t} $longscore $longmember
+                for {set i 1} {$i < $count} {incr i} {
+                    r zadd zero-source{t} -0 [format m%03d $i]
+                }
+                set zero_expected [r zrange zero-source{t} 0 -1 withscores]
+                set payload [r dump zero-source{t}]
+                r config set zset-max-listpack-entries 128
+                r restore zero-restored{t} 0 $payload
+                assert_encoding btree zero-restored{t}
+                assert_equal $zero_expected [r zrange zero-restored{t} 0 -1 withscores]
+                assert_equal $payload [r dump zero-restored{t}]
+            }
+        }
+        r config set zset-max-listpack-entries $old_limit
+        r del zero-source{t} zero-restored{t}
+    }
+
+    test {Small ZSET_2 loads still select listpack after all members are known} {
+        set old_limit [lindex [r config get zset-max-listpack-entries] 1]
+        r del small-load-source{t} small-load-restored{t}
+        r config set zset-max-listpack-entries 0
+        r zadd small-load-source{t} 1 a 2 b 3 c
+        set payload [r dump small-load-source{t}]
+        r config set zset-max-listpack-entries 128
+        r restore small-load-restored{t} 0 $payload
+        assert_encoding listpack small-load-restored{t}
+        assert_equal {a 1 b 2 c 3} [r zrange small-load-restored{t} 0 -1 withscores]
+        r config set zset-max-listpack-entries $old_limit
+        r del small-load-source{t} small-load-restored{t}
+    }
+
     # A helper function to verify either ZPOP* or ZMPOP* response.
     proc verify_pop_response {pop res zpop_expected_response zmpop_expected_response} {
         if {[string match "*ZM*" $pop]} {
@@ -102,7 +140,7 @@ start_server {tags {"zset"}} {
         if {$encoding == "listpack"} {
             r config set zset-max-ziplist-entries 128
             r config set zset-max-ziplist-value 64
-        } elseif {$encoding == "skiplist"} {
+        } elseif {$encoding == "btree"} {
             r config set zset-max-ziplist-entries 0
             r config set zset-max-ziplist-value 0
         } else {
@@ -550,6 +588,10 @@ start_server {tags {"zset"}} {
             assert_equal {f e d} [r zrevrangebyscore zset 6 3]
             assert_equal {g f e} [r zrevrangebyscore zset +inf 4]
             assert_equal 3 [r zcount zset 0 3]
+            assert_equal 7 [r zcount zset -inf +inf]
+            assert_equal 6 [r zcount zset (-inf +inf]
+            assert_equal 6 [r zcount zset -inf (+inf]
+            assert_equal 5 [r zcount zset (-inf (+inf]
 
             # exclusive range
             assert_equal {b}   [r zrangebyscore zset (-inf (2]
@@ -1231,6 +1273,18 @@ start_server {tags {"zset"}} {
             }
         }
 
+    test {ZPOPMIN keeps mixed signed zero scores exact in a B+tree leaf} {
+        set old_limit [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del signed-zero-pop
+        r zadd signed-zero-pop 0 a 0 b 0 c -0 d
+        assert_encoding btree signed-zero-pop
+        assert_equal {a 0} [r zpopmin signed-zero-pop]
+        assert_equal {b 0 c 0 d -0} [r zrange signed-zero-pop 0 -1 withscores]
+        r del signed-zero-pop
+        r config set zset-max-listpack-entries $old_limit
+    }
+
     foreach {popmin popmax} {ZPOPMIN ZPOPMAX ZMPOP_MIN ZMPOP_MAX} {
         test "Basic $popmin/$popmax with a single key - $encoding" {
             r del zset
@@ -1338,7 +1392,192 @@ start_server {tags {"zset"}} {
     }
 
     basics listpack
-    basics skiplist
+    basics btree
+
+    test {ZSET with a large member stays in the B+ tree encoding} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del large-member-zset{t}
+        r zadd large-member-zset{t} 1 short
+        assert_encoding btree large-member-zset{t}
+
+        set large [string repeat x 4096]
+        r zadd large-member-zset{t} 2 $large
+        assert_encoding btree large-member-zset{t}
+        assert_equal {short 1} [r zrange large-member-zset{t} 0 0 withscores]
+        assert_equal 2 [r zscore large-member-zset{t} $large]
+
+        # Exercise lookup, ordering, score changes, and scans while the long
+        # member is stored outside its score leaf.
+        r zadd large-member-zset{t} 1.5 middle 3 tail
+        assert_equal 2 [r zrank large-member-zset{t} $large]
+        assert_equal 1 [r zrevrank large-member-zset{t} $large]
+        assert_equal {middle 1.5} \
+            [r zrange large-member-zset{t} 1.5 1.5 byscore withscores]
+        assert_equal 0 [r zadd large-member-zset{t} 0 $large]
+        assert_equal $large [r zrange large-member-zset{t} 0 0]
+
+        set scanned [lindex [r zscan large-member-zset{t} 0 count 100] 1]
+        set scanned_members {}
+        foreach {member score} $scanned {
+            lappend scanned_members $member
+        }
+        assert_equal [lsort [list short middle tail $large]] \
+                     [lsort $scanned_members]
+
+        r del large-member-other{t}
+        r zadd large-member-other{t} 5 $large 7 other
+        assert_encoding btree large-member-other{t}
+        assert_equal [list $large 5] \
+            [r zinter 2 large-member-zset{t} large-member-other{t} withscores]
+
+        r del large-member-copy{t}
+        assert_equal 1 [r copy large-member-zset{t} large-member-copy{t}]
+        assert_encoding btree large-member-copy{t}
+        assert_equal [r zrange large-member-zset{t} 0 -1 withscores] \
+                     [r zrange large-member-copy{t} 0 -1 withscores]
+        assert_equal 1 [r zrem large-member-zset{t} middle]
+        assert_equal 1 [r zremrangebyscore large-member-zset{t} 3 3]
+        assert_equal [list $large 0] [r zpopmin large-member-zset{t}]
+        assert_equal 0 [r zscore large-member-copy{t} $large]
+
+        r config set zset-max-listpack-entries $original_max
+    }
+
+    test {ZSET B+ tree handles members at the external storage boundary} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del member-storage-boundary
+
+        set members {}
+        for {set j 0} {$j < 400} {incr j} {
+            set len [lindex {319 320 321} [expr {$j % 3}]]
+            set prefix [format "%04d:" $j]
+            set member "$prefix[string repeat x [expr {$len-[string length $prefix]}]]"
+            assert_equal $len [string length $member]
+            lappend members $member
+            r zadd member-storage-boundary $j $member
+        }
+        assert_encoding btree member-storage-boundary
+        assert_equal 400 [r zcard member-storage-boundary]
+
+        # Move inline and external members across many leaves, then remove
+        # ranges which force leaves to be rebuilt and joined.
+        for {set j 0} {$j < 400} {incr j 17} {
+            assert_equal 0 [r zadd member-storage-boundary [expr {1000-$j}] \
+                                [lindex $members $j]]
+        }
+        assert_equal 1000 [r zscore member-storage-boundary [lindex $members 0]]
+        assert_equal 1 [r zrem member-storage-boundary [lindex $members 319]]
+        assert_equal 100 [r zremrangebyrank member-storage-boundary 50 149]
+        assert_equal 299 [r zcard member-storage-boundary]
+
+        foreach j {1 2 202 398 399} {
+            set member [lindex $members $j]
+            if {$j != 319} {
+                assert_equal $j [r zscore member-storage-boundary $member]
+            }
+        }
+
+        r config set zset-max-listpack-entries $original_max
+    }
+
+    test {RDB preserves a B+ tree containing a large member} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del large-member-rdb{t}
+        set large [string repeat x 4096]
+        r zadd large-member-rdb{t} 1 short 2 $large
+        assert_encoding btree large-member-rdb{t}
+
+        r debug reload
+        assert_encoding btree large-member-rdb{t}
+        assert_equal [list short 1 $large 2] \
+                     [r zrange large-member-rdb{t} 0 -1 withscores]
+        r config set zset-max-listpack-entries $original_max
+    } {OK} {needs:debug}
+
+    test {ZSET btree byte-full edge pages preserve order} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del edge-right{t} edge-left{t}
+
+        for {set j 0} {$j < 200} {incr j} {
+            set member [format "%04d%s" $j [string repeat x 296]]
+            r zadd edge-right{t} $j $member
+            r zadd edge-left{t} [expr {-$j}] $member
+        }
+
+        set first [format "%04d%s" 0 [string repeat x 296]]
+        set last [format "%04d%s" 199 [string repeat x 296]]
+        assert_encoding btree edge-right{t}
+        assert_encoding btree edge-left{t}
+        assert_equal $first [r zrange edge-right{t} 0 0]
+        assert_equal $last [r zrange edge-right{t} -1 -1]
+        assert_equal $last [r zrange edge-left{t} 0 0]
+        assert_equal $first [r zrange edge-left{t} -1 -1]
+
+        assert_equal 150 [r zremrangebyrank edge-right{t} 25 174]
+        assert_equal [r zrange edge-right{t} 0 -1] \
+                     [lreverse [r zrevrange edge-right{t} 0 -1]]
+
+        r config set zset-max-listpack-entries $original_max
+    }
+
+    test {ZSET btree large range deletion rebuilds a usable member index} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del delete-rank{t} delete-score{t}
+
+        set args {}
+        for {set j 0} {$j < 2048} {incr j} {
+            set member [format "member:%04d:%s" $j [string repeat x 372]]
+            lappend args $j $member
+        }
+        assert_equal 2048 [r zadd delete-rank{t} {*}$args]
+        assert_equal 1 [r copy delete-rank{t} delete-score{t}]
+
+        assert_equal 1025 [r zremrangebyrank delete-rank{t} 512 1536]
+        assert_equal 1025 [r zremrangebyscore delete-score{t} 512 1536]
+        assert_equal [r zrange delete-rank{t} 0 -1 withscores] \
+                     [r zrange delete-score{t} 0 -1 withscores]
+        assert_equal 1023 [r zcard delete-rank{t}]
+        set member511 [format "member:%04d:%s" 511 [string repeat x 372]]
+        set member512 [format "member:%04d:%s" 512 [string repeat x 372]]
+        set member1537 [format "member:%04d:%s" 1537 [string repeat x 372]]
+        assert_equal 511 [r zrank delete-rank{t} $member511]
+        assert_equal 512 [r zrank delete-rank{t} $member1537]
+        assert_equal {} [r zscore delete-rank{t} $member512]
+
+        set member0 [format "member:%04d:%s" 0 [string repeat x 372]]
+        set member2047 [format "member:%04d:%s" 2047 [string repeat x 372]]
+        foreach key {delete-rank{t} delete-score{t}} {
+            assert_equal 1 [r zadd $key 1000 new-member]
+            assert_equal 0 [r zadd $key 3000 $member0]
+            assert_equal 1 [r zrem $key $member2047]
+        }
+        assert_equal [r zrange delete-rank{t} 0 -1 withscores] \
+                     [r zrange delete-score{t} 0 -1 withscores]
+
+        r config set zset-max-listpack-entries $original_max
+    }
+
+    test {ZSET btree range deletion can empty a medium-sized index} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 0
+        r del delete-all
+
+        set args {}
+        for {set j 0} {$j < 500} {incr j} {
+            lappend args $j member:$j
+        }
+        assert_equal 500 [r zadd delete-all {*}$args]
+        assert_encoding btree delete-all
+        assert_equal 500 [r zremrangebyrank delete-all 0 -1]
+        assert_equal 0 [r exists delete-all]
+
+        r config set zset-max-listpack-entries $original_max
+    }
 
     test "ZPOP/ZMPOP against wrong type" {
         r set foo{t} bar
@@ -1634,6 +1873,56 @@ start_server {tags {"zset"}} {
         }
     }
 
+    test {ZUNIONSTORE btree-encoded result matches aggregate-max with score collisions and weights} {
+        # Result is forced past zset-max-listpack-entries so it lands on the
+        # BTREE bulk-build fast path, with heavy score collisions (only 50
+        # distinct scores across thousands of members) to exercise the
+        # (score, member) tie-break in the same order the B+tree itself
+        # expects.
+        set orig_listpack_entries [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 8
+        r del u1{t} u2{t} udest{t}
+        array set expected {}
+        set cmd1 [list r zadd u1{t}]
+        for {set j 0} {$j < 300} {incr j} {
+            set m [format "m%05d" $j]
+            set sc [expr {$j % 20}]
+            lappend cmd1 $sc $m
+            set expected($m) $sc
+        }
+        {*}$cmd1
+        set cmd2 [list r zadd u2{t}]
+        for {set j 150} {$j < 450} {incr j} {
+            set m [format "m%05d" $j]
+            set sc [expr {($j % 20) * 2}]
+            lappend cmd2 $sc $m
+            if {[info exists expected($m)]} {
+                if {$sc > $expected($m)} { set expected($m) $sc }
+            } else {
+                set expected($m) $sc
+            }
+        }
+        {*}$cmd2
+        r zunionstore udest{t} 2 u1{t} u2{t} aggregate max
+        assert_encoding btree udest{t}
+        assert_equal [array size expected] [r zcard udest{t}]
+        set prevscore {}
+        set prevele {}
+        foreach {ele score} [r zrange udest{t} 0 -1 withscores] {
+            assert_equal $expected($ele) $score
+            if {$prevscore ne {}} {
+                if {$score == $prevscore} {
+                    assert {[string compare $ele $prevele] >= 0}
+                } else {
+                    assert {$score >= $prevscore}
+                }
+            }
+            set prevscore $score
+            set prevele $ele
+        }
+        r config set zset-max-listpack-entries $orig_listpack_entries
+    }
+
     test "ZUNIONSTORE/ZINTERSTORE/ZDIFFSTORE error if using WITHSCORES " {
         assert_error "*ERR*syntax*" {r zunionstore foo{t} 2 zsetd{t} zsetf{t} withscores}
         assert_error "*ERR*syntax*" {r zinterstore foo{t} 2 zsetd{t} zsetf{t} withscores}
@@ -1678,6 +1967,32 @@ start_server {tags {"zset"}} {
         assert_match {*ERR*wrong*number*arg*} $e
     }
 
+    test {ZMSCORE B+tree batches preserve order, duplicates and missing members} {
+        r del mscore-batches
+        set batch_args {}
+        for {set i 0} {$i < 300} {incr i} {
+            lappend batch_args [expr {$i - 150}] member:$i
+        }
+        r zadd mscore-batches {*}$batch_args
+        assert_encoding btree mscore-batches
+        foreach count {1 2 15 16 17 31 32 33 63 64 65 100} {
+            set batch_members {}
+            set batch_scores {}
+            for {set i 0} {$i < $count} {incr i} {
+                if {$i % 3 == 0} {
+                    lappend batch_members missing:$i
+                    lappend batch_scores {}
+                } else {
+                    set member [expr {($i * 37) % 23}]
+                    lappend batch_members member:$member
+                    lappend batch_scores [expr {$member - 150}]
+                }
+            }
+            assert_equal $batch_scores [r zmscore mscore-batches {*}$batch_members]
+        }
+        r del mscore-batches
+    }
+
     test "ZSET commands don't accept the empty strings as valid score" {
         assert_error "*not*float*" {r zadd myzset "" abc}
     }
@@ -1700,7 +2015,7 @@ start_server {tags {"zset"}} {
             r config set zset-max-ziplist-entries 256
             r config set zset-max-ziplist-value 64
             set elements 128
-        } elseif {$encoding == "skiplist"} {
+        } elseif {$encoding == "btree"} {
             r config set zset-max-ziplist-entries 0
             r config set zset-max-ziplist-value 0
             if {$::accurate} {set elements 1000} else {set elements 100}
@@ -1764,7 +2079,7 @@ start_server {tags {"zset"}} {
         test "ZSCORE 17-19 significant digit mantissas (widened fast path) - $encoding" {
             # Exercise the widened fast_float_strtod path that handles
             # mantissas > 2^53 (via __uint128_t arithmetic). ZADD/ZSCORE
-            # must round-trip bit-exactly through the listpack/skiplist
+            # must round-trip bit-exactly through both zset encodings
             # encoding (parse on ingest, parse again on retrieval). Each
             # input string below parses to a specific IEEE double whose
             # canonical string representation is itself, so `expr` in Tcl
@@ -2028,7 +2343,7 @@ start_server {tags {"zset"}} {
             }
         }
 
-        test "ZSETs skiplist implementation backlink consistency test - $encoding" {
+        test "ZSET forward/reverse consistency test - $encoding" {
             set diff 0
             for {set j 0} {$j < $elements} {incr j} {
                 r zadd myzset [expr rand()] "Element-$j"
@@ -2046,7 +2361,7 @@ start_server {tags {"zset"}} {
             assert_equal 0 $diff
         }
 
-        test "ZSETs ZRANK augmented skip list stress testing - $encoding" {
+        test "ZSET ZRANK stress testing - $encoding" {
             set err {}
             r del myzset
             for {set k 0} {$k < 2000} {incr k} {
@@ -2214,7 +2529,7 @@ start_server {tags {"zset"}} {
 
     tags {"slow"} {
         stresses listpack
-        stresses skiplist
+        stresses btree
     }
 
     test "BZPOP/BZMPOP against wrong type" {
@@ -2362,7 +2677,7 @@ start_server {tags {"zset"}} {
         $rd2 close
     } {0} {cluster:skip}
 
-    test {ZSET skiplist order consistency when elements are moved} {
+    test {ZSET order consistency when elements are moved} {
         set original_max [lindex [r config get zset-max-ziplist-entries] 1]
         r config set zset-max-ziplist-entries 0
         for {set times 0} {$times < 10} {incr times} {
@@ -2489,13 +2804,13 @@ start_server {tags {"zset"}} {
         r config set zset-max-listpack-entries 0
         r del z1{t} z2{t}
         r zadd z1{t} 1 a
-        assert_encoding skiplist z1{t}
+        assert_encoding btree z1{t}
         assert_equal 1 [r zrangestore z2{t} z1{t} 0 -1]
-        assert_encoding skiplist z2{t}
+        assert_encoding btree z2{t}
         r config set zset-max-listpack-entries $original_max
     }
 
-    test {ZRANGESTORE with zset-max-listpack-entries 1 dst key should use skiplist encoding} {
+    test {ZRANGESTORE with zset-max-listpack-entries 1 dst key should use btree encoding} {
         set original_max [lindex [r config get zset-max-listpack-entries] 1]
         r config set zset-max-listpack-entries 1
         r del z1{t} z2{t} z3{t}
@@ -2503,7 +2818,151 @@ start_server {tags {"zset"}} {
         assert_equal 1 [r zrangestore z2{t} z1{t} 0 0]
         assert_encoding listpack z2{t}
         assert_equal 2 [r zrangestore z3{t} z1{t} 0 1]
-        assert_encoding skiplist z3{t}
+        assert_encoding btree z3{t}
+        r config set zset-max-listpack-entries $original_max
+    }
+
+    test {ZADD converts an existing listpack after lowering its limit} {
+        set conversion_limit [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 128
+        r del conversion{t}
+        set conversion_args {}
+        for {set i 0} {$i < 40} {incr i} {
+            lappend conversion_args $i m$i
+        }
+        r zadd conversion{t} {*}$conversion_args
+        assert_encoding listpack conversion{t}
+        r config set zset-max-listpack-entries 1
+        assert_equal 2 [r zadd conversion{t} 1 x 2 y]
+        assert_encoding btree conversion{t}
+        assert_equal 42 [r zcard conversion{t}]
+        foreach {score member} $conversion_args {
+            assert_equal $score [r zscore conversion{t} $member]
+        }
+        r del conversion{t}
+        r config set zset-max-listpack-entries $conversion_limit
+    }
+
+    test {ZRANGESTORE bulk append during member-index rehash} {
+        set old_entries [lindex [r config get zset-max-listpack-entries] 1]
+        set old_value [lindex [r config get zset-max-listpack-value] 1]
+        r config set zset-max-listpack-entries 128
+        r config set zset-max-listpack-value 64
+        foreach size {65 127 319 320 4096} {
+            foreach count {40 60 100 1000} {
+                r del rehash-src{t} rehash-dst{t}
+                set elements {}
+                for {set j 0} {$j < $count} {incr j} {
+                    set member [format %08d $j][string repeat x [expr {$size - 8}]]
+                    lappend elements $j $member
+                }
+                r zadd rehash-src{t} {*}$elements
+                set rehash_expected [r zrange rehash-src{t} 0 -1 withscores]
+                foreach bounds {{0 -1} {-inf +inf BYSCORE}} {
+                    assert_equal $count [r zrangestore rehash-dst{t} rehash-src{t} {*}$bounds]
+                    assert_encoding btree rehash-dst{t}
+                    assert_equal $rehash_expected [r zrange rehash-dst{t} 0 -1 withscores]
+                    # Point reads advance and finish any remaining rehash.
+                    foreach {member score} $rehash_expected {
+                        assert_equal $score [r zscore rehash-dst{t} $member]
+                    }
+                }
+            }
+        }
+        r del rehash-src{t} rehash-dst{t}
+        r config set zset-max-listpack-entries $old_entries
+        r config set zset-max-listpack-value $old_value
+    }
+
+    test {ZRANGESTORE btree-encoded result matches expected slice, forward and REV, with score collisions} {
+        set original_max [lindex [r config get zset-max-listpack-entries] 1]
+        r config set zset-max-listpack-entries 8
+        r del zsrc{t} zdst_fwd{t} zdst_rev{t} zdst_byscore{t} zdst_byscore_rev{t} zdst_lim{t}
+        set cmd [list r zadd zsrc{t}]
+        array set expected {}
+        for {set j 0} {$j < 500} {incr j} {
+            set m [format "member:%04d" $j]
+            set sc [expr {$j % 20}]
+            lappend cmd $sc $m
+            set expected($m) $sc
+        }
+        {*}$cmd
+        assert_encoding btree zsrc{t}
+
+        # Ground truth: the full (score, member) ascending order, matching
+        # both zrangestore's own selection order and the B+tree's own order.
+        set full_sorted {}
+        foreach {ele score} [r zrange zsrc{t} 0 -1 withscores] {
+            lappend full_sorted [list $score $ele]
+        }
+        assert_equal 500 [llength $full_sorted]
+
+        # Forward rank slice -- exercises the new descent-free append fast
+        # path directly (ascending direction, BTREE destination).
+        r zrangestore zdst_fwd{t} zsrc{t} 100 399
+        assert_encoding btree zdst_fwd{t}
+        set got_fwd {}
+        foreach {ele score} [r zrange zdst_fwd{t} 0 -1 withscores] {
+            lappend got_fwd [list $score $ele]
+        }
+        assert_equal [lrange $full_sorted 100 399] $got_fwd
+
+        # REV rank slice -- must keep using the original (slower but always
+        # correct) zsetAdd() path; the fast path is guarded off for reverse
+        # emission. Rank indices for REV count from the highest score down.
+        r zrangestore zdst_rev{t} zsrc{t} 100 399 REV
+        assert_encoding btree zdst_rev{t}
+        set full_desc [lreverse $full_sorted]
+        set expected_rev [lsort -command {apply {{a b} {
+            set sa [lindex $a 0]; set sb [lindex $b 0]
+            if {$sa != $sb} { return [expr {$sa < $sb ? -1 : 1}] }
+            return [string compare [lindex $a 1] [lindex $b 1]]
+        }}} [lrange $full_desc 100 399]]
+        set got_rev {}
+        foreach {ele score} [r zrange zdst_rev{t} 0 -1 withscores] {
+            lappend got_rev [list $score $ele]
+        }
+        assert_equal $expected_rev $got_rev
+
+        # BYSCORE forward -- same fast path, different selection method
+        # feeding the same STORE result handler.
+        r zrangestore zdst_byscore{t} zsrc{t} 3 7 BYSCORE
+        assert_encoding btree zdst_byscore{t}
+        set expected_byscore {}
+        foreach pair $full_sorted {
+            set sc [lindex $pair 0]
+            if {$sc >= 3 && $sc <= 7} { lappend expected_byscore $pair }
+        }
+        set got_byscore {}
+        foreach {ele score} [r zrange zdst_byscore{t} 0 -1 withscores] {
+            lappend got_byscore [list $score $ele]
+        }
+        assert_equal $expected_byscore $got_byscore
+
+        # BYSCORE REV -- fast path must NOT engage here either.
+        r zrangestore zdst_byscore_rev{t} zsrc{t} 7 3 BYSCORE REV
+        assert_encoding btree zdst_byscore_rev{t}
+        set got_byscore_rev {}
+        foreach {ele score} [r zrange zdst_byscore_rev{t} 0 -1 withscores] {
+            lappend got_byscore_rev [list $score $ele]
+        }
+        assert_equal $expected_byscore $got_byscore_rev
+
+        # BYSCORE with LIMIT offset/count -- forward, still fast-path eligible.
+        r zrangestore zdst_lim{t} zsrc{t} 0 19 BYSCORE LIMIT 10 50
+        assert_encoding btree zdst_lim{t}
+        set expected_lim {}
+        foreach pair $full_sorted {
+            set sc [lindex $pair 0]
+            if {$sc >= 0 && $sc <= 19} { lappend expected_lim $pair }
+        }
+        set expected_lim [lrange $expected_lim 10 59]
+        set got_lim {}
+        foreach {ele score} [r zrange zdst_lim{t} 0 -1 withscores] {
+            lappend got_lim [list $score $ele]
+        }
+        assert_equal $expected_lim $got_lim
+
         r config set zset-max-listpack-entries $original_max
     }
 
@@ -2540,7 +2999,7 @@ start_server {tags {"zset"}} {
         }
     }
 
-    foreach {type contents} "listpack {1 a 2 b 3 c} skiplist {1 a 2 b 3 [randstring 70 90 alpha]}" {
+    foreach {type contents} "listpack {1 a 2 b 3 c} btree {1 a 2 b 3 [randstring 70 90 alpha]}" {
         set original_max_value [lindex [r config get zset-max-ziplist-value] 1]
         r config set zset-max-ziplist-value 10
         create_zset myzset $contents
@@ -2599,7 +3058,7 @@ start_server {tags {"zset"}} {
     r readraw 0
 
     foreach {type contents} "
-        skiplist {1 a 2 b 3 c 4 d 5 e 6 f 7 g 7 h 9 i 10 [randstring 70 90 alpha]}
+        btree {1 a 2 b 3 c 4 d 5 e 6 f 7 g 7 h 9 i 10 [randstring 70 90 alpha]}
         listpack {1 a 2 b 3 c 4 d 5 e 6 f 7 g 7 h 9 i 10 j} " {
         test "ZRANDMEMBER with <count> - $type" {
             set original_max_value [lindex [r config get zset-max-ziplist-value] 1]
@@ -2776,7 +3235,7 @@ start_server {tags {"zset"}} {
                 assert_encoding hashtable set_big{t}
             }
 
-            foreach zset_type {listpack skiplist} {
+            foreach zset_type {listpack btree} {
                 r del zset_small{t} zset_big{t}
 
                 if {$zset_type == "listpack"} {
@@ -2784,12 +3243,12 @@ start_server {tags {"zset"}} {
                     r zadd zset_big{t} 1 1 2 2 3 3 4 4 5 5
                     assert_encoding listpack zset_small{t}
                     assert_encoding listpack zset_big{t}
-                } elseif {$zset_type == "skiplist"} {
+                } elseif {$zset_type == "btree"} {
                     r config set zset-max-listpack-entries 0
                     r zadd zset_small{t} 1 1 2 2 3 3
                     r zadd zset_big{t} 1 1 2 2 3 3 4 4 5 5
-                    assert_encoding skiplist zset_small{t}
-                    assert_encoding skiplist zset_big{t}
+                    assert_encoding btree zset_small{t}
+                    assert_encoding btree zset_big{t}
                 }
 
                 # Test one key is big and one key is small separately.
@@ -2850,9 +3309,98 @@ start_server {tags {"zset"}} {
             assert_encoding listpack myzset
             assert_equal $max_entries [r zcard myzset]
             assert_equal 1 [r zadd myzset 1 b]
-            assert_encoding skiplist myzset
+            assert_encoding btree myzset
+            assert_equal [expr {$max_entries + 1}] [r zcard myzset]
+            assert {[r zscore myzset 1] ne {}}
 
             r config set zset-max-listpack-entries $original_max
         }
     }
+
+    if {$::debug_defrag} {
+        test "Active defrag relocates all B+ tree zset pages" {
+            r config set activedefrag no
+            r config set hz 100
+            r config set active-defrag-threshold-lower 0
+            r config set active-defrag-threshold-upper 1
+            r config set active-defrag-ignore-bytes 1
+            r config set active-defrag-cycle-min 75
+            r config set active-defrag-cycle-max 75
+            r config set active-defrag-max-scan-fields 10
+            r config set zset-max-listpack-entries 0
+            r del defrag-zset
+
+            # The last insertion starts an index resize and leaves both hash
+            # table allocations alive for the forced defrag pass.
+            set elements 7937
+            set args {}
+            for {set i 0} {$i < $elements} {incr i} {
+                set member [format "member:%05d" $i]
+                if {$i >= 5000 && $i % 100 == 0} {
+                    append member [string repeat x 512]
+                }
+                lappend args $i $member
+            }
+            assert_equal $elements [r zadd defrag-zset {*}$args]
+            assert_encoding btree defrag-zset
+            set digest [debug_digest]
+
+            r config resetstat
+            r config set activedefrag yes
+            wait_for_condition 1000 10 {
+                [s active_defrag_hits] > 1000
+            } else {
+                fail "Active defrag did not relocate the B+ tree pages"
+            }
+            r config set activedefrag no
+            wait_for_condition 1000 10 {
+                [s active_defrag_running] == 0
+            } else {
+                fail "Active defrag did not stop"
+            }
+
+            assert_equal $elements [r zcard defrag-zset]
+            assert_equal 1234 [r zscore defrag-zset member:01234]
+            assert_equal 1234 [r zrank defrag-zset member:01234]
+            assert_equal {member:00000 member:00001 member:00002} \
+                [r zrange defrag-zset 0 2]
+
+            # Exercise the tree and member index after their allocations move.
+            assert_equal 0 [r zadd defrag-zset 99999 member:01234]
+            assert_equal [expr {$elements - 1}] \
+                         [r zrank defrag-zset member:01234]
+            assert_equal 0 [r zadd defrag-zset 1234 member:01234]
+            assert_equal 1 [r zrem defrag-zset member:04321]
+            assert_equal 1 [r zadd defrag-zset 4321 member:04321]
+            assert_equal $digest [debug_digest]
+            assert_equal OK [r save]
+
+            # Small trees are moved immediately instead of being queued.
+            r del defrag-zset
+            r config set active-defrag-max-scan-fields 100
+            assert_equal 5 [r zadd defrag-zset 1 one 2 two 3 three 4 four 5 five]
+            assert_encoding btree defrag-zset
+            set digest [debug_digest]
+
+            r config resetstat
+            r config set activedefrag yes
+            wait_for_condition 1000 10 {
+                [s active_defrag_hits] > 4
+            } else {
+                fail "Active defrag did not relocate the small B+ tree"
+            }
+            r config set activedefrag no
+            wait_for_condition 1000 10 {
+                [s active_defrag_running] == 0
+            } else {
+                fail "Active defrag did not stop"
+            }
+
+            assert_equal {one two three four five} \
+                [r zrange defrag-zset 0 -1]
+            assert_equal 3 [r zscore defrag-zset three]
+            assert_equal $digest [debug_digest]
+            assert_equal OK [r save]
+        }
+    }
 }