Fix for ZUNIONSTORE bug when there is an empty set among input sets. Regression test...

[redis.git] / src / t_zset.c

diff --git a/src/t_zset.c b/src/t_zset.c
index 63657890974343f860017d76a80264368a07e953..19aac3d50a3792cfa0bb4d0ce3839039466bde61 100644 (file)
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -174,12 +174,6 @@ int zslDelete(zskiplist *zsl, double score, robj *obj) {
     return 0; /* not found */
 }
 
-/* Struct to hold a inclusive/exclusive range spec. */
-typedef struct {
-    double min, max;
-    int minex, maxex; /* are min or max exclusive? */
-} zrangespec;
-
 static int zslValueGteMin(double value, zrangespec *spec) {
     return spec->minex ? (value > spec->min) : (value >= spec->min);
 }
@@ -616,7 +610,7 @@ unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, do
     unsigned char *sptr;
     char scorebuf[128];
     int scorelen;
-    int offset;
+    size_t offset;
 
     redisAssert(ele->encoding == REDIS_ENCODING_RAW);
     scorelen = d2string(scorebuf,sizeof(scorebuf),score);
@@ -721,7 +715,7 @@ unsigned int zsetLength(robj *zobj) {
     int length = -1;
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         length = zzlLength(zobj->ptr);
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         length = ((zset*)zobj->ptr)->zsl->length;
     } else {
         redisPanic("Unknown sorted set encoding");
@@ -743,7 +737,7 @@ void zsetConvert(robj *zobj, int encoding) {
         unsigned int vlen;
         long long vlong;
 
-        if (encoding != REDIS_ENCODING_RAW)
+        if (encoding != REDIS_ENCODING_SKIPLIST)
             redisPanic("Unknown target encoding");
 
         zs = zmalloc(sizeof(*zs));
@@ -772,8 +766,8 @@ void zsetConvert(robj *zobj, int encoding) {
 
         zfree(zobj->ptr);
         zobj->ptr = zs;
-        zobj->encoding = REDIS_ENCODING_RAW;
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        zobj->encoding = REDIS_ENCODING_SKIPLIST;
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         unsigned char *zl = ziplistNew();
 
         if (encoding != REDIS_ENCODING_ZIPLIST)
@@ -872,9 +866,9 @@ void zaddGenericCommand(redisClient *c, int incr) {
              * too long *before* executing zzlInsert. */
             zobj->ptr = zzlInsert(zobj->ptr,ele,score);
             if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
-                zsetConvert(zobj,REDIS_ENCODING_RAW);
+                zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
             if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
-                zsetConvert(zobj,REDIS_ENCODING_RAW);
+                zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
 
             signalModifiedKey(c->db,key);
             server.dirty++;
@@ -884,7 +878,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
             else /* ZADD */
                 addReply(c,shared.cone);
         }
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         zskiplistNode *znode;
         dictEntry *de;
@@ -967,7 +961,7 @@ void zremCommand(redisClient *c) {
             addReply(c,shared.czero);
             return;
         }
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         dictEntry *de;
         double score;
@@ -1012,7 +1006,8 @@ void zremrangebyscoreCommand(redisClient *c) {
 
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
         if (htNeedsResize(zs->dict)) dictResize(zs->dict);
@@ -1057,7 +1052,8 @@ void zremrangebyrankCommand(redisClient *c) {
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         /* Correct for 1-based rank. */
         zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
 
         /* Correct for 1-based rank. */
@@ -1157,7 +1153,7 @@ void zuiInitIterator(zsetopsrc *op) {
                 it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
                 redisAssert(it->zl.sptr != NULL);
             }
-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
             it->sl.zs = op->subject->ptr;
             it->sl.node = it->sl.zs->zsl->header->level[0].forward;
         } else {
@@ -1185,7 +1181,7 @@ void zuiClearIterator(zsetopsrc *op) {
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             REDIS_NOTUSED(it); /* skip */
-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
             REDIS_NOTUSED(it); /* skip */
         } else {
             redisPanic("Unknown sorted set encoding");
@@ -1212,7 +1208,7 @@ int zuiLength(zsetopsrc *op) {
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             return zzlLength(it->zl.zl);
-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
             return it->sl.zs->zsl->length;
         } else {
             redisPanic("Unknown sorted set encoding");
@@ -1237,7 +1233,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
     if (op->type == REDIS_SET) {
         iterset *it = &op->iter.set;
         if (op->encoding == REDIS_ENCODING_INTSET) {
-            if (!intsetGet(it->is.is,it->is.ii,&val->ell))
+            if (!intsetGet(it->is.is,it->is.ii,(int64_t*)&val->ell))
                 return 0;
             val->score = 1.0;
 
@@ -1265,7 +1261,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
 
             /* Move to next element. */
             zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
             if (it->sl.node == NULL)
                 return 0;
             val->ele = it->sl.node->obj;
@@ -1377,7 +1373,7 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
             } else {
                 return 0;
             }
-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
             dictEntry *de;
             if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
                 *score = *(double*)dictGetEntryVal(de);
@@ -1513,6 +1509,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
     dstobj = createZsetObject();
     dstzset = dstobj->ptr;
+    memset(&zval, 0, sizeof(zval));
 
     if (op == REDIS_OP_INTER) {
         /* Skip everything if the smallest input is empty. */
@@ -1524,7 +1521,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
                 score = src[0].weight * zval.score;
                 for (j = 1; j < setnum; j++) {
-                    if (zuiFind(&src[j],&zval,&value)) {
+                    /* It is not safe to access the hash we zset we are
+                     * iterating, so explicitly check for equal object. */
+                    if (src[j].subject == src[0].subject) {
+                        value = zval.score*src[j].weight;
+                        zunionInterAggregate(&score,value,aggregate);
+                    } else if (zuiFind(&src[j],&zval,&value)) {
                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     } else {
@@ -1548,7 +1550,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
         }
     } else if (op == REDIS_OP_UNION) {
         for (i = 0; i < setnum; i++) {
-            if (zuiLength(&src[0]) == 0)
+            if (zuiLength(&src[i]) == 0)
                 continue;
 
             while (zuiNext(&src[i],&zval)) {
@@ -1564,7 +1566,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                 /* Because the inputs are sorted by size, it's only possible
                  * for sets at larger indices to hold this element. */
                 for (j = (i+1); j < setnum; j++) {
-                    if (zuiFind(&src[j],&zval,&value)) {
+                    /* It is not safe to access the hash we zset we are
+                     * iterating, so explicitly check for equal object. */
+                    if(src[j].subject == src[i].subject) {
+                        value = zval.score*src[j].weight;
+                        zunionInterAggregate(&score,value,aggregate);
+                    } else if (zuiFind(&src[j],&zval,&value)) {
                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     }
@@ -1690,7 +1697,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
                 zzlNext(zl,&eptr,&sptr);
         }
 
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;
@@ -1847,7 +1854,7 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) {
             else
                 zzlNext(zl,&eptr,&sptr);
         }
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;
@@ -1941,7 +1948,7 @@ void zscoreCommand(redisClient *c) {
             addReplyDouble(c,score);
         else
             addReply(c,shared.nullbulk);
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         dictEntry *de;
 
@@ -1995,7 +2002,7 @@ void zrankGenericCommand(redisClient *c, int reverse) {
         } else {
             addReply(c,shared.nullbulk);
         }
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         dictEntry *de;