X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/8218db3dae220409f239a5a0adb47d106b178bf9..4c5f0966b2e582981d9fdaf3b511c6cf4ac4d4d5:/src/t_zset.c

diff --git a/src/t_zset.c b/src/t_zset.c
index 1aab4b12..8665aa14 100644
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -449,11 +449,130 @@ int zzlCompareElements(unsigned char *eptr, unsigned char *cstr, unsigned int cl
     return cmp;
 }
 
-unsigned int *zzlLength(robj *zobj) {
+unsigned int zzlLength(robj *zobj) {
     unsigned char *zl = zobj->ptr;
     return ziplistLen(zl)/2;
 }
 
+/* Move to next entry based on the values in eptr and sptr. Both are set to
+ * NULL when there is no next entry. */
+void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
+    unsigned char *_eptr, *_sptr;
+    redisAssert(*eptr != NULL && *sptr != NULL);
+
+    _eptr = ziplistNext(zl,*sptr);
+    if (_eptr != NULL) {
+        _sptr = ziplistNext(zl,_eptr);
+        redisAssert(_sptr != NULL);
+    } else {
+        /* No next entry. */
+        _sptr = NULL;
+    }
+
+    *eptr = _eptr;
+    *sptr = _sptr;
+}
+
+/* Move to the previous entry based on the values in eptr and sptr. Both are
+ * set to NULL when there is no next entry. */
+void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
+    unsigned char *_eptr, *_sptr;
+    redisAssert(*eptr != NULL && *sptr != NULL);
+
+    _sptr = ziplistPrev(zl,*eptr);
+    if (_sptr != NULL) {
+        _eptr = ziplistPrev(zl,_sptr);
+        redisAssert(_eptr != NULL);
+    } else {
+        /* No previous entry. */
+        _eptr = NULL;
+    }
+
+    *eptr = _eptr;
+    *sptr = _sptr;
+}
+
+/* Returns if there is a part of the zset is in range. Should only be used
+ * internally by zzlFirstInRange and zzlLastInRange. */
+int zzlIsInRange(unsigned char *zl, zrangespec *range) {
+    unsigned char *p;
+    double score;
+
+    /* Test for ranges that will always be empty. */
+    if (range->min > range->max ||
+            (range->min == range->max && (range->minex || range->maxex)))
+        return 0;
+
+    p = ziplistIndex(zl,-1); /* Last score. */
+    redisAssert(p != NULL);
+    score = zzlGetScore(p);
+    if (!zslValueGteMin(score,range))
+        return 0;
+
+    p = ziplistIndex(zl,1); /* First score. */
+    redisAssert(p != NULL);
+    score = zzlGetScore(p);
+    if (!zslValueLteMax(score,range))
+        return 0;
+
+    return 1;
+}
+
+/* Find pointer to the first element contained in the specified range.
+ * Returns NULL when no element is contained in the range. */
+unsigned char *zzlFirstInRange(robj *zobj, zrangespec range) {
+    unsigned char *zl = zobj->ptr;
+    unsigned char *eptr = ziplistIndex(zl,0), *sptr;
+    double score;
+
+    /* If everything is out of range, return early. */
+    if (!zzlIsInRange(zl,&range)) return NULL;
+
+    while (eptr != NULL) {
+        sptr = ziplistNext(zl,eptr);
+        redisAssert(sptr != NULL);
+
+        score = zzlGetScore(sptr);
+        if (zslValueGteMin(score,&range))
+            return eptr;
+
+        /* Move to next element. */
+        eptr = ziplistNext(zl,sptr);
+    }
+
+    return NULL;
+}
+
+/* Find pointer to the last element contained in the specified range.
+ * Returns NULL when no element is contained in the range. */
+unsigned char *zzlLastInRange(robj *zobj, zrangespec range) {
+    unsigned char *zl = zobj->ptr;
+    unsigned char *eptr = ziplistIndex(zl,-2), *sptr;
+    double score;
+
+    /* If everything is out of range, return early. */
+    if (!zzlIsInRange(zl,&range)) return NULL;
+
+    while (eptr != NULL) {
+        sptr = ziplistNext(zl,eptr);
+        redisAssert(sptr != NULL);
+
+        score = zzlGetScore(sptr);
+        if (zslValueLteMax(score,&range))
+            return eptr;
+
+        /* Move to previous element by moving to the score of previous element.
+         * When this returns NULL, we know there also is no element. */
+        sptr = ziplistPrev(zl,eptr);
+        if (sptr != NULL)
+            redisAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
+        else
+            eptr = NULL;
+    }
+
+    return NULL;
+}
+
 unsigned char *zzlFind(robj *zobj, robj *ele, double *score) {
     unsigned char *zl = zobj->ptr;
     unsigned char *eptr = ziplistIndex(zl,0), *sptr;
@@ -557,6 +676,58 @@ int zzlInsert(robj *zobj, robj *ele, double score) {
     return REDIS_OK;
 }
 
+unsigned long zzlDeleteRangeByScore(robj *zobj, zrangespec range) {
+    unsigned char *zl = zobj->ptr;
+    unsigned char *eptr, *sptr;
+    double score;
+    unsigned long deleted = 0;
+
+    eptr = zzlFirstInRange(zobj,range);
+    if (eptr == NULL) return deleted;
+
+
+    /* When the tail of the ziplist is deleted, eptr will point to the sentinel
+     * byte and ziplistNext will return NULL. */
+    while ((sptr = ziplistNext(zl,eptr)) != NULL) {
+        score = zzlGetScore(sptr);
+        if (zslValueLteMax(score,&range)) {
+            /* Delete both the element and the score. */
+            zl = ziplistDelete(zl,&eptr);
+            zl = ziplistDelete(zl,&eptr);
+            deleted++;
+        } else {
+            /* No longer in range. */
+            break;
+        }
+    }
+
+    return deleted;
+}
+
+/* Delete all the elements with rank between start and end from the skiplist.
+ * Start and end are inclusive. Note that start and end need to be 1-based */
+unsigned long zzlDeleteRangeByRank(robj *zobj, unsigned int start, unsigned int end) {
+    unsigned int num = (end-start)+1;
+    zobj->ptr = ziplistDeleteRange(zobj->ptr,2*(start-1),2*num);
+    return num;
+}
+
+/*-----------------------------------------------------------------------------
+ * Common sorted set API
+ *----------------------------------------------------------------------------*/
+
+int zsLength(robj *zobj) {
+    int length = -1;
+    if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
+        length = zzlLength(zobj);
+    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        length = ((zset*)zobj->ptr)->zsl->length;
+    } else {
+        redisPanic("Unknown sorted set encoding");
+    }
+    return length;
+}
+
 /*-----------------------------------------------------------------------------
  * Sorted set commands 
  *----------------------------------------------------------------------------*/
@@ -736,10 +907,10 @@ void zremCommand(redisClient *c) {
 }
 
 void zremrangebyscoreCommand(redisClient *c) {
+    robj *key = c->argv[1];
+    robj *zobj;
     zrangespec range;
-    long deleted;
-    robj *o;
-    zset *zs;
+    unsigned long deleted;
 
     /* Parse the range arguments. */
     if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
@@ -747,35 +918,41 @@ void zremrangebyscoreCommand(redisClient *c) {
         return;
     }
 
-    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
-        checkType(c,o,REDIS_ZSET)) return;
+    if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
+        checkType(c,zobj,REDIS_ZSET)) return;
 
-    zs = o->ptr;
-    deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
-    if (htNeedsResize(zs->dict)) dictResize(zs->dict);
-    if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
-    if (deleted) signalModifiedKey(c->db,c->argv[1]);
+    if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
+        deleted = zzlDeleteRangeByScore(zobj,range);
+    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        zset *zs = zobj->ptr;
+        deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
+    } else {
+        redisPanic("Unknown sorted set encoding");
+    }
+
+    if (deleted) signalModifiedKey(c->db,key);
     server.dirty += deleted;
     addReplyLongLong(c,deleted);
 }
 
 void zremrangebyrankCommand(redisClient *c) {
+    robj *key = c->argv[1];
+    robj *zobj;
     long start;
     long end;
     int llen;
-    long deleted;
-    robj *zsetobj;
-    zset *zs;
+    unsigned long deleted;
 
     if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
         (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
 
-    if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
-        checkType(c,zsetobj,REDIS_ZSET)) return;
-    zs = zsetobj->ptr;
-    llen = zs->zsl->length;
+    if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
+        checkType(c,zobj,REDIS_ZSET)) return;
 
-    /* convert negative indexes */
+    /* Sanitize indexes. */
+    llen = zsLength(zobj);
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
@@ -788,14 +965,23 @@ void zremrangebyrankCommand(redisClient *c) {
     }
     if (end >= llen) end = llen-1;
 
-    /* increment start and end because zsl*Rank functions
-     * use 1-based rank */
-    deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
-    if (htNeedsResize(zs->dict)) dictResize(zs->dict);
-    if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
-    if (deleted) signalModifiedKey(c->db,c->argv[1]);
+    if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
+        /* Correct for 1-based rank. */
+        deleted = zzlDeleteRangeByRank(zobj,start+1,end+1);
+    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        zset *zs = zobj->ptr;
+
+        /* Correct for 1-based rank. */
+        deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
+    } else {
+        redisPanic("Unknown sorted set encoding");
+    }
+
+    if (deleted) signalModifiedKey(c->db,key);
     server.dirty += deleted;
-    addReplyLongLong(c, deleted);
+    addReplyLongLong(c,deleted);
 }
 
 typedef struct {
@@ -1019,16 +1205,13 @@ void zinterstoreCommand(redisClient *c) {
 }
 
 void zrangeGenericCommand(redisClient *c, int reverse) {
-    robj *o;
+    robj *key = c->argv[1];
+    robj *zobj;
+    int withscores = 0;
     long start;
     long end;
-    int withscores = 0;
     int llen;
-    int rangelen, j;
-    zset *zsetobj;
-    zskiplist *zsl;
-    zskiplistNode *ln;
-    robj *ele;
+    int rangelen;
 
     if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
         (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
@@ -1040,13 +1223,11 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
         return;
     }
 
-    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
-         || checkType(c,o,REDIS_ZSET)) return;
-    zsetobj = o->ptr;
-    zsl = zsetobj->zsl;
-    llen = zsl->length;
+    if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL
+         || checkType(c,zobj,REDIS_ZSET)) return;
 
-    /* convert negative indexes */
+    /* Sanitize indexes. */
+    llen = zsLength(zobj);
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
@@ -1060,23 +1241,68 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
     if (end >= llen) end = llen-1;
     rangelen = (end-start)+1;
 
-    /* check if starting point is trivial, before searching
-     * the element in log(N) time */
-    if (reverse) {
-        ln = start == 0 ? zsl->tail : zslGetElementByRank(zsl, llen-start);
-    } else {
-        ln = start == 0 ?
-            zsl->header->level[0].forward : zslGetElementByRank(zsl, start+1);
-    }
-
     /* Return the result in form of a multi-bulk reply */
-    addReplyMultiBulkLen(c,withscores ? (rangelen*2) : rangelen);
-    for (j = 0; j < rangelen; j++) {
-        ele = ln->obj;
-        addReplyBulk(c,ele);
-        if (withscores)
-            addReplyDouble(c,ln->score);
-        ln = reverse ? ln->backward : ln->level[0].forward;
+    addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen);
+
+    if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
+        unsigned char *zl = zobj->ptr;
+        unsigned char *eptr, *sptr;
+        unsigned char *vstr;
+        unsigned int vlen;
+        long long vlong;
+
+        if (reverse)
+            eptr = ziplistIndex(zl,-2-(2*start));
+        else
+            eptr = ziplistIndex(zl,2*start);
+
+        redisAssert(eptr != NULL);
+        sptr = ziplistNext(zl,eptr);
+
+        while (rangelen--) {
+            redisAssert(eptr != NULL && sptr != NULL);
+            redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
+            if (vstr == NULL)
+                addReplyBulkLongLong(c,vlong);
+            else
+                addReplyBulkCBuffer(c,vstr,vlen);
+
+            if (withscores)
+                addReplyDouble(c,zzlGetScore(sptr));
+
+            if (reverse)
+                zzlPrev(zl,&eptr,&sptr);
+            else
+                zzlNext(zl,&eptr,&sptr);
+        }
+
+    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        zset *zs = zobj->ptr;
+        zskiplist *zsl = zs->zsl;
+        zskiplistNode *ln;
+        robj *ele;
+
+        /* Check if starting point is trivial, before doing log(N) lookup. */
+        if (reverse) {
+            ln = zsl->tail;
+            if (start > 0)
+                ln = zslGetElementByRank(zsl,llen-start);
+        } else {
+            ln = zsl->header->level[0].forward;
+            if (start > 0)
+                ln = zslGetElementByRank(zsl,start+1);
+        }
+
+        while(rangelen--) {
+            redisAssert(ln != NULL);
+            ele = ln->obj;
+            addReplyBulk(c,ele);
+            if (withscores)
+                addReplyDouble(c,ln->score);
+            ln = reverse ? ln->backward : ln->level[0].forward;
+        }
+    } else {
+        redisPanic("Unknown sorted set encoding");
     }
 }