]> git.saurik.com Git - redis.git/blame - src/t_zset.c
Don't assume that "char" is signed.
[redis.git] / src / t_zset.c
CommitLineData
e2641e09 1#include "redis.h"
2
3#include <math.h>
4
5/*-----------------------------------------------------------------------------
6 * Sorted set API
7 *----------------------------------------------------------------------------*/
8
9/* ZSETs are ordered sets using two data structures to hold the same elements
10 * in order to get O(log(N)) INSERT and REMOVE operations into a sorted
11 * data structure.
12 *
13 * The elements are added to an hash table mapping Redis objects to scores.
14 * At the same time the elements are added to a skip list mapping scores
15 * to Redis objects (so objects are sorted by scores in this "view"). */
16
17/* This skiplist implementation is almost a C translation of the original
18 * algorithm described by William Pugh in "Skip Lists: A Probabilistic
19 * Alternative to Balanced Trees", modified in three ways:
d8bd12f9 20 * a) this implementation allows for repeated scores.
e2641e09 21 * b) the comparison is not just by key (our 'score') but by satellite data.
22 * c) there is a back pointer, so it's a doubly linked list with the back
23 * pointers being only at "level 1". This allows to traverse the list
24 * from tail to head, useful for ZREVRANGE. */
25
26zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
2159782b 27 zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
e2641e09 28 zn->score = score;
29 zn->obj = obj;
30 return zn;
31}
32
33zskiplist *zslCreate(void) {
34 int j;
35 zskiplist *zsl;
36
37 zsl = zmalloc(sizeof(*zsl));
38 zsl->level = 1;
39 zsl->length = 0;
40 zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
41 for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
2159782b
PN
42 zsl->header->level[j].forward = NULL;
43 zsl->header->level[j].span = 0;
e2641e09 44 }
45 zsl->header->backward = NULL;
46 zsl->tail = NULL;
47 return zsl;
48}
49
50void zslFreeNode(zskiplistNode *node) {
51 decrRefCount(node->obj);
e2641e09 52 zfree(node);
53}
54
55void zslFree(zskiplist *zsl) {
2159782b 56 zskiplistNode *node = zsl->header->level[0].forward, *next;
e2641e09 57
e2641e09 58 zfree(zsl->header);
59 while(node) {
2159782b 60 next = node->level[0].forward;
e2641e09 61 zslFreeNode(node);
62 node = next;
63 }
64 zfree(zsl);
65}
66
7faa1f07 67/* Returns a random level for the new skiplist node we are going to create.
68 * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL
69 * (both inclusive), with a powerlaw-alike distribution where higher
70 * levels are less likely to be returned. */
e2641e09 71int zslRandomLevel(void) {
72 int level = 1;
73 while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
74 level += 1;
75 return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
76}
77
69ef89f2 78zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
e2641e09 79 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
80 unsigned int rank[ZSKIPLIST_MAXLEVEL];
81 int i, level;
82
a244a13b 83 redisAssert(!isnan(score));
e2641e09 84 x = zsl->header;
85 for (i = zsl->level-1; i >= 0; i--) {
86 /* store rank that is crossed to reach the insert position */
87 rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
2159782b
PN
88 while (x->level[i].forward &&
89 (x->level[i].forward->score < score ||
90 (x->level[i].forward->score == score &&
91 compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
92 rank[i] += x->level[i].span;
93 x = x->level[i].forward;
e2641e09 94 }
95 update[i] = x;
96 }
97 /* we assume the key is not already inside, since we allow duplicated
98 * scores, and the re-insertion of score and redis object should never
99 * happpen since the caller of zslInsert() should test in the hash table
100 * if the element is already inside or not. */
101 level = zslRandomLevel();
102 if (level > zsl->level) {
103 for (i = zsl->level; i < level; i++) {
104 rank[i] = 0;
105 update[i] = zsl->header;
2159782b 106 update[i]->level[i].span = zsl->length;
e2641e09 107 }
108 zsl->level = level;
109 }
110 x = zslCreateNode(level,score,obj);
111 for (i = 0; i < level; i++) {
2159782b
PN
112 x->level[i].forward = update[i]->level[i].forward;
113 update[i]->level[i].forward = x;
e2641e09 114
115 /* update span covered by update[i] as x is inserted here */
2159782b
PN
116 x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
117 update[i]->level[i].span = (rank[0] - rank[i]) + 1;
e2641e09 118 }
119
120 /* increment span for untouched levels */
121 for (i = level; i < zsl->level; i++) {
2159782b 122 update[i]->level[i].span++;
e2641e09 123 }
124
125 x->backward = (update[0] == zsl->header) ? NULL : update[0];
2159782b
PN
126 if (x->level[0].forward)
127 x->level[0].forward->backward = x;
e2641e09 128 else
129 zsl->tail = x;
130 zsl->length++;
69ef89f2 131 return x;
e2641e09 132}
133
134/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
135void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
136 int i;
137 for (i = 0; i < zsl->level; i++) {
2159782b
PN
138 if (update[i]->level[i].forward == x) {
139 update[i]->level[i].span += x->level[i].span - 1;
140 update[i]->level[i].forward = x->level[i].forward;
e2641e09 141 } else {
2159782b 142 update[i]->level[i].span -= 1;
e2641e09 143 }
144 }
2159782b
PN
145 if (x->level[0].forward) {
146 x->level[0].forward->backward = x->backward;
e2641e09 147 } else {
148 zsl->tail = x->backward;
149 }
2159782b 150 while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
e2641e09 151 zsl->level--;
152 zsl->length--;
153}
154
155/* Delete an element with matching score/object from the skiplist. */
156int zslDelete(zskiplist *zsl, double score, robj *obj) {
157 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
158 int i;
159
160 x = zsl->header;
161 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
162 while (x->level[i].forward &&
163 (x->level[i].forward->score < score ||
164 (x->level[i].forward->score == score &&
165 compareStringObjects(x->level[i].forward->obj,obj) < 0)))
166 x = x->level[i].forward;
e2641e09 167 update[i] = x;
168 }
169 /* We may have multiple elements with the same score, what we need
170 * is to find the element with both the right score and object. */
2159782b 171 x = x->level[0].forward;
e2641e09 172 if (x && score == x->score && equalStringObjects(x->obj,obj)) {
173 zslDeleteNode(zsl, x, update);
174 zslFreeNode(x);
175 return 1;
176 } else {
177 return 0; /* not found */
178 }
179 return 0; /* not found */
180}
181
45290ad9 182static int zslValueGteMin(double value, zrangespec *spec) {
22b9bf15
PN
183 return spec->minex ? (value > spec->min) : (value >= spec->min);
184}
185
45290ad9 186static int zslValueLteMax(double value, zrangespec *spec) {
22b9bf15
PN
187 return spec->maxex ? (value < spec->max) : (value <= spec->max);
188}
189
22b9bf15
PN
190/* Returns if there is a part of the zset is in range. */
191int zslIsInRange(zskiplist *zsl, zrangespec *range) {
192 zskiplistNode *x;
193
8e1b3277
PN
194 /* Test for ranges that will always be empty. */
195 if (range->min > range->max ||
196 (range->min == range->max && (range->minex || range->maxex)))
197 return 0;
22b9bf15 198 x = zsl->tail;
45290ad9 199 if (x == NULL || !zslValueGteMin(x->score,range))
22b9bf15
PN
200 return 0;
201 x = zsl->header->level[0].forward;
45290ad9 202 if (x == NULL || !zslValueLteMax(x->score,range))
22b9bf15
PN
203 return 0;
204 return 1;
205}
206
207/* Find the first node that is contained in the specified range.
208 * Returns NULL when no element is contained in the range. */
209zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) {
210 zskiplistNode *x;
211 int i;
212
213 /* If everything is out of range, return early. */
214 if (!zslIsInRange(zsl,&range)) return NULL;
215
216 x = zsl->header;
217 for (i = zsl->level-1; i >= 0; i--) {
218 /* Go forward while *OUT* of range. */
219 while (x->level[i].forward &&
45290ad9 220 !zslValueGteMin(x->level[i].forward->score,&range))
22b9bf15
PN
221 x = x->level[i].forward;
222 }
223
e53ca04b 224 /* This is an inner range, so the next node cannot be NULL. */
22b9bf15 225 x = x->level[0].forward;
e53ca04b
PN
226 redisAssert(x != NULL);
227
228 /* Check if score <= max. */
229 if (!zslValueLteMax(x->score,&range)) return NULL;
22b9bf15
PN
230 return x;
231}
232
233/* Find the last node that is contained in the specified range.
234 * Returns NULL when no element is contained in the range. */
235zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) {
236 zskiplistNode *x;
237 int i;
238
239 /* If everything is out of range, return early. */
240 if (!zslIsInRange(zsl,&range)) return NULL;
241
242 x = zsl->header;
243 for (i = zsl->level-1; i >= 0; i--) {
244 /* Go forward while *IN* range. */
245 while (x->level[i].forward &&
45290ad9 246 zslValueLteMax(x->level[i].forward->score,&range))
22b9bf15
PN
247 x = x->level[i].forward;
248 }
249
e53ca04b
PN
250 /* This is an inner range, so this node cannot be NULL. */
251 redisAssert(x != NULL);
252
253 /* Check if score >= min. */
254 if (!zslValueGteMin(x->score,&range)) return NULL;
22b9bf15
PN
255 return x;
256}
257
e2641e09 258/* Delete all the elements with score between min and max from the skiplist.
259 * Min and mx are inclusive, so a score >= min || score <= max is deleted.
260 * Note that this function takes the reference to the hash table view of the
261 * sorted set, in order to remove the elements from the hash table too. */
91504b6c 262unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec range, dict *dict) {
e2641e09 263 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
264 unsigned long removed = 0;
265 int i;
266
267 x = zsl->header;
268 for (i = zsl->level-1; i >= 0; i--) {
91504b6c
PN
269 while (x->level[i].forward && (range.minex ?
270 x->level[i].forward->score <= range.min :
271 x->level[i].forward->score < range.min))
272 x = x->level[i].forward;
e2641e09 273 update[i] = x;
274 }
91504b6c
PN
275
276 /* Current node is the last with score < or <= min. */
2159782b 277 x = x->level[0].forward;
91504b6c
PN
278
279 /* Delete nodes while in range. */
280 while (x && (range.maxex ? x->score < range.max : x->score <= range.max)) {
2159782b 281 zskiplistNode *next = x->level[0].forward;
69ef89f2 282 zslDeleteNode(zsl,x,update);
e2641e09 283 dictDelete(dict,x->obj);
284 zslFreeNode(x);
285 removed++;
286 x = next;
287 }
91504b6c 288 return removed;
e2641e09 289}
290
291/* Delete all the elements with rank between start and end from the skiplist.
292 * Start and end are inclusive. Note that start and end need to be 1-based */
293unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
294 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
295 unsigned long traversed = 0, removed = 0;
296 int i;
297
298 x = zsl->header;
299 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
300 while (x->level[i].forward && (traversed + x->level[i].span) < start) {
301 traversed += x->level[i].span;
302 x = x->level[i].forward;
e2641e09 303 }
304 update[i] = x;
305 }
306
307 traversed++;
2159782b 308 x = x->level[0].forward;
e2641e09 309 while (x && traversed <= end) {
2159782b 310 zskiplistNode *next = x->level[0].forward;
69ef89f2 311 zslDeleteNode(zsl,x,update);
e2641e09 312 dictDelete(dict,x->obj);
313 zslFreeNode(x);
314 removed++;
315 traversed++;
316 x = next;
317 }
318 return removed;
319}
320
e2641e09 321/* Find the rank for an element by both score and key.
322 * Returns 0 when the element cannot be found, rank otherwise.
323 * Note that the rank is 1-based due to the span of zsl->header to the
324 * first element. */
a3004773 325unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
e2641e09 326 zskiplistNode *x;
327 unsigned long rank = 0;
328 int i;
329
330 x = zsl->header;
331 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
332 while (x->level[i].forward &&
333 (x->level[i].forward->score < score ||
334 (x->level[i].forward->score == score &&
335 compareStringObjects(x->level[i].forward->obj,o) <= 0))) {
336 rank += x->level[i].span;
337 x = x->level[i].forward;
e2641e09 338 }
339
340 /* x might be equal to zsl->header, so test if obj is non-NULL */
341 if (x->obj && equalStringObjects(x->obj,o)) {
342 return rank;
343 }
344 }
345 return 0;
346}
347
348/* Finds an element by its rank. The rank argument needs to be 1-based. */
a3004773 349zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
e2641e09 350 zskiplistNode *x;
351 unsigned long traversed = 0;
352 int i;
353
354 x = zsl->header;
355 for (i = zsl->level-1; i >= 0; i--) {
2159782b 356 while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
e2641e09 357 {
2159782b
PN
358 traversed += x->level[i].span;
359 x = x->level[i].forward;
e2641e09 360 }
361 if (traversed == rank) {
362 return x;
363 }
364 }
365 return NULL;
366}
367
25bb8a44 368/* Populate the rangespec according to the objects min and max. */
7236fdb2
PN
369static int zslParseRange(robj *min, robj *max, zrangespec *spec) {
370 char *eptr;
25bb8a44
PN
371 spec->minex = spec->maxex = 0;
372
373 /* Parse the min-max interval. If one of the values is prefixed
374 * by the "(" character, it's considered "open". For instance
375 * ZRANGEBYSCORE zset (1.5 (2.5 will match min < x < max
376 * ZRANGEBYSCORE zset 1.5 2.5 will instead match min <= x <= max */
377 if (min->encoding == REDIS_ENCODING_INT) {
378 spec->min = (long)min->ptr;
379 } else {
380 if (((char*)min->ptr)[0] == '(') {
7236fdb2
PN
381 spec->min = strtod((char*)min->ptr+1,&eptr);
382 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR;
25bb8a44
PN
383 spec->minex = 1;
384 } else {
7236fdb2
PN
385 spec->min = strtod((char*)min->ptr,&eptr);
386 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR;
25bb8a44
PN
387 }
388 }
389 if (max->encoding == REDIS_ENCODING_INT) {
390 spec->max = (long)max->ptr;
391 } else {
392 if (((char*)max->ptr)[0] == '(') {
7236fdb2
PN
393 spec->max = strtod((char*)max->ptr+1,&eptr);
394 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR;
25bb8a44
PN
395 spec->maxex = 1;
396 } else {
7236fdb2
PN
397 spec->max = strtod((char*)max->ptr,&eptr);
398 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR;
25bb8a44
PN
399 }
400 }
401
402 return REDIS_OK;
403}
404
21c5b508
PN
405/*-----------------------------------------------------------------------------
406 * Ziplist-backed sorted set API
407 *----------------------------------------------------------------------------*/
408
409double zzlGetScore(unsigned char *sptr) {
410 unsigned char *vstr;
411 unsigned int vlen;
412 long long vlong;
413 char buf[128];
414 double score;
415
416 redisAssert(sptr != NULL);
417 redisAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
418
419 if (vstr) {
420 memcpy(buf,vstr,vlen);
421 buf[vlen] = '\0';
422 score = strtod(buf,NULL);
423 } else {
424 score = vlong;
425 }
426
427 return score;
428}
429
430/* Compare element in sorted set with given element. */
431int zzlCompareElements(unsigned char *eptr, unsigned char *cstr, unsigned int clen) {
432 unsigned char *vstr;
433 unsigned int vlen;
434 long long vlong;
435 unsigned char vbuf[32];
436 int minlen, cmp;
437
438 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
439 if (vstr == NULL) {
440 /* Store string representation of long long in buf. */
441 vlen = ll2string((char*)vbuf,sizeof(vbuf),vlong);
442 vstr = vbuf;
443 }
444
445 minlen = (vlen < clen) ? vlen : clen;
446 cmp = memcmp(vstr,cstr,minlen);
447 if (cmp == 0) return vlen-clen;
448 return cmp;
449}
450
bbfe232f 451unsigned int zzlLength(unsigned char *zl) {
0b10e104
PN
452 return ziplistLen(zl)/2;
453}
454
4c5f0966
PN
455/* Move to next entry based on the values in eptr and sptr. Both are set to
456 * NULL when there is no next entry. */
457void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
458 unsigned char *_eptr, *_sptr;
459 redisAssert(*eptr != NULL && *sptr != NULL);
460
461 _eptr = ziplistNext(zl,*sptr);
462 if (_eptr != NULL) {
463 _sptr = ziplistNext(zl,_eptr);
464 redisAssert(_sptr != NULL);
465 } else {
466 /* No next entry. */
467 _sptr = NULL;
468 }
469
470 *eptr = _eptr;
471 *sptr = _sptr;
472}
473
474/* Move to the previous entry based on the values in eptr and sptr. Both are
475 * set to NULL when there is no next entry. */
476void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
477 unsigned char *_eptr, *_sptr;
478 redisAssert(*eptr != NULL && *sptr != NULL);
479
480 _sptr = ziplistPrev(zl,*eptr);
481 if (_sptr != NULL) {
482 _eptr = ziplistPrev(zl,_sptr);
483 redisAssert(_eptr != NULL);
484 } else {
485 /* No previous entry. */
486 _eptr = NULL;
487 }
488
489 *eptr = _eptr;
490 *sptr = _sptr;
491}
492
4a14dbba
PN
493/* Returns if there is a part of the zset is in range. Should only be used
494 * internally by zzlFirstInRange and zzlLastInRange. */
495int zzlIsInRange(unsigned char *zl, zrangespec *range) {
496 unsigned char *p;
497 double score;
498
499 /* Test for ranges that will always be empty. */
500 if (range->min > range->max ||
501 (range->min == range->max && (range->minex || range->maxex)))
502 return 0;
503
504 p = ziplistIndex(zl,-1); /* Last score. */
feb28288 505 if (p == NULL) return 0; /* Empty sorted set */
4a14dbba
PN
506 score = zzlGetScore(p);
507 if (!zslValueGteMin(score,range))
508 return 0;
509
510 p = ziplistIndex(zl,1); /* First score. */
511 redisAssert(p != NULL);
512 score = zzlGetScore(p);
513 if (!zslValueLteMax(score,range))
514 return 0;
515
516 return 1;
517}
518
519/* Find pointer to the first element contained in the specified range.
520 * Returns NULL when no element is contained in the range. */
8588bfa3 521unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) {
4a14dbba
PN
522 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
523 double score;
524
525 /* If everything is out of range, return early. */
526 if (!zzlIsInRange(zl,&range)) return NULL;
527
528 while (eptr != NULL) {
529 sptr = ziplistNext(zl,eptr);
530 redisAssert(sptr != NULL);
531
532 score = zzlGetScore(sptr);
e53ca04b
PN
533 if (zslValueGteMin(score,&range)) {
534 /* Check if score <= max. */
535 if (zslValueLteMax(score,&range))
536 return eptr;
537 return NULL;
538 }
4a14dbba
PN
539
540 /* Move to next element. */
541 eptr = ziplistNext(zl,sptr);
542 }
543
544 return NULL;
545}
546
547/* Find pointer to the last element contained in the specified range.
548 * Returns NULL when no element is contained in the range. */
8588bfa3 549unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) {
4a14dbba
PN
550 unsigned char *eptr = ziplistIndex(zl,-2), *sptr;
551 double score;
552
553 /* If everything is out of range, return early. */
554 if (!zzlIsInRange(zl,&range)) return NULL;
555
556 while (eptr != NULL) {
557 sptr = ziplistNext(zl,eptr);
558 redisAssert(sptr != NULL);
559
560 score = zzlGetScore(sptr);
e53ca04b
PN
561 if (zslValueLteMax(score,&range)) {
562 /* Check if score >= min. */
563 if (zslValueGteMin(score,&range))
564 return eptr;
565 return NULL;
566 }
4a14dbba
PN
567
568 /* Move to previous element by moving to the score of previous element.
569 * When this returns NULL, we know there also is no element. */
570 sptr = ziplistPrev(zl,eptr);
571 if (sptr != NULL)
572 redisAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
573 else
574 eptr = NULL;
575 }
576
577 return NULL;
578}
579
8588bfa3 580unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) {
21c5b508
PN
581 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
582
583 ele = getDecodedObject(ele);
584 while (eptr != NULL) {
585 sptr = ziplistNext(zl,eptr);
eab0e26e 586 redisAssertWithInfo(NULL,ele,sptr != NULL);
21c5b508
PN
587
588 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) {
589 /* Matching element, pull out score. */
0b10e104 590 if (score != NULL) *score = zzlGetScore(sptr);
21c5b508
PN
591 decrRefCount(ele);
592 return eptr;
593 }
594
595 /* Move to next element. */
596 eptr = ziplistNext(zl,sptr);
597 }
598
599 decrRefCount(ele);
600 return NULL;
601}
602
603/* Delete (element,score) pair from ziplist. Use local copy of eptr because we
604 * don't want to modify the one given as argument. */
8588bfa3 605unsigned char *zzlDelete(unsigned char *zl, unsigned char *eptr) {
21c5b508
PN
606 unsigned char *p = eptr;
607
608 /* TODO: add function to ziplist API to delete N elements from offset. */
609 zl = ziplistDelete(zl,&p);
610 zl = ziplistDelete(zl,&p);
8588bfa3 611 return zl;
21c5b508
PN
612}
613
8588bfa3 614unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score) {
21c5b508
PN
615 unsigned char *sptr;
616 char scorebuf[128];
617 int scorelen;
69298a05 618 size_t offset;
21c5b508 619
eab0e26e 620 redisAssertWithInfo(NULL,ele,ele->encoding == REDIS_ENCODING_RAW);
21c5b508
PN
621 scorelen = d2string(scorebuf,sizeof(scorebuf),score);
622 if (eptr == NULL) {
623 zl = ziplistPush(zl,ele->ptr,sdslen(ele->ptr),ZIPLIST_TAIL);
624 zl = ziplistPush(zl,(unsigned char*)scorebuf,scorelen,ZIPLIST_TAIL);
625 } else {
626 /* Keep offset relative to zl, as it might be re-allocated. */
627 offset = eptr-zl;
628 zl = ziplistInsert(zl,eptr,ele->ptr,sdslen(ele->ptr));
629 eptr = zl+offset;
630
631 /* Insert score after the element. */
eab0e26e 632 redisAssertWithInfo(NULL,ele,(sptr = ziplistNext(zl,eptr)) != NULL);
21c5b508
PN
633 zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen);
634 }
635
8588bfa3 636 return zl;
21c5b508
PN
637}
638
639/* Insert (element,score) pair in ziplist. This function assumes the element is
640 * not yet present in the list. */
8588bfa3 641unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) {
21c5b508
PN
642 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
643 double s;
21c5b508
PN
644
645 ele = getDecodedObject(ele);
646 while (eptr != NULL) {
647 sptr = ziplistNext(zl,eptr);
eab0e26e 648 redisAssertWithInfo(NULL,ele,sptr != NULL);
21c5b508
PN
649 s = zzlGetScore(sptr);
650
651 if (s > score) {
652 /* First element with score larger than score for element to be
653 * inserted. This means we should take its spot in the list to
654 * maintain ordering. */
8588bfa3 655 zl = zzlInsertAt(zl,eptr,ele,score);
21c5b508 656 break;
8218db3d
PN
657 } else if (s == score) {
658 /* Ensure lexicographical ordering for elements. */
d1c920c5 659 if (zzlCompareElements(eptr,ele->ptr,sdslen(ele->ptr)) > 0) {
8588bfa3 660 zl = zzlInsertAt(zl,eptr,ele,score);
8218db3d
PN
661 break;
662 }
21c5b508
PN
663 }
664
665 /* Move to next element. */
666 eptr = ziplistNext(zl,sptr);
667 }
668
669 /* Push on tail of list when it was not yet inserted. */
8218db3d 670 if (eptr == NULL)
8588bfa3 671 zl = zzlInsertAt(zl,NULL,ele,score);
21c5b508
PN
672
673 decrRefCount(ele);
8588bfa3 674 return zl;
21c5b508 675}
25bb8a44 676
8588bfa3 677unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec range, unsigned long *deleted) {
4a14dbba
PN
678 unsigned char *eptr, *sptr;
679 double score;
8588bfa3 680 unsigned long num = 0;
4a14dbba 681
8588bfa3 682 if (deleted != NULL) *deleted = 0;
4a14dbba 683
8588bfa3
PN
684 eptr = zzlFirstInRange(zl,range);
685 if (eptr == NULL) return zl;
4a14dbba
PN
686
687 /* When the tail of the ziplist is deleted, eptr will point to the sentinel
688 * byte and ziplistNext will return NULL. */
689 while ((sptr = ziplistNext(zl,eptr)) != NULL) {
690 score = zzlGetScore(sptr);
691 if (zslValueLteMax(score,&range)) {
692 /* Delete both the element and the score. */
693 zl = ziplistDelete(zl,&eptr);
694 zl = ziplistDelete(zl,&eptr);
8588bfa3 695 num++;
4a14dbba
PN
696 } else {
697 /* No longer in range. */
698 break;
699 }
700 }
701
8588bfa3
PN
702 if (deleted != NULL) *deleted = num;
703 return zl;
4a14dbba
PN
704}
705
63b7b7fb
PN
706/* Delete all the elements with rank between start and end from the skiplist.
707 * Start and end are inclusive. Note that start and end need to be 1-based */
8588bfa3 708unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted) {
63b7b7fb 709 unsigned int num = (end-start)+1;
8588bfa3
PN
710 if (deleted) *deleted = num;
711 zl = ziplistDeleteRange(zl,2*(start-1),2*num);
712 return zl;
63b7b7fb
PN
713}
714
5d1b4fb6
PN
715/*-----------------------------------------------------------------------------
716 * Common sorted set API
717 *----------------------------------------------------------------------------*/
718
df26a0ae 719unsigned int zsetLength(robj *zobj) {
5d1b4fb6
PN
720 int length = -1;
721 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
bbfe232f 722 length = zzlLength(zobj->ptr);
100ed062 723 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
5d1b4fb6
PN
724 length = ((zset*)zobj->ptr)->zsl->length;
725 } else {
726 redisPanic("Unknown sorted set encoding");
727 }
728 return length;
729}
730
df26a0ae 731void zsetConvert(robj *zobj, int encoding) {
a669d5e9
PN
732 zset *zs;
733 zskiplistNode *node, *next;
734 robj *ele;
735 double score;
736
737 if (zobj->encoding == encoding) return;
738 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
739 unsigned char *zl = zobj->ptr;
740 unsigned char *eptr, *sptr;
741 unsigned char *vstr;
742 unsigned int vlen;
743 long long vlong;
744
100ed062 745 if (encoding != REDIS_ENCODING_SKIPLIST)
a669d5e9
PN
746 redisPanic("Unknown target encoding");
747
748 zs = zmalloc(sizeof(*zs));
749 zs->dict = dictCreate(&zsetDictType,NULL);
750 zs->zsl = zslCreate();
751
752 eptr = ziplistIndex(zl,0);
eab0e26e 753 redisAssertWithInfo(NULL,zobj,eptr != NULL);
a669d5e9 754 sptr = ziplistNext(zl,eptr);
eab0e26e 755 redisAssertWithInfo(NULL,zobj,sptr != NULL);
a669d5e9
PN
756
757 while (eptr != NULL) {
758 score = zzlGetScore(sptr);
eab0e26e 759 redisAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
a669d5e9
PN
760 if (vstr == NULL)
761 ele = createStringObjectFromLongLong(vlong);
762 else
763 ele = createStringObject((char*)vstr,vlen);
764
765 /* Has incremented refcount since it was just created. */
766 node = zslInsert(zs->zsl,score,ele);
eab0e26e 767 redisAssertWithInfo(NULL,zobj,dictAdd(zs->dict,ele,&node->score) == DICT_OK);
a669d5e9
PN
768 incrRefCount(ele); /* Added to dictionary. */
769 zzlNext(zl,&eptr,&sptr);
770 }
771
772 zfree(zobj->ptr);
773 zobj->ptr = zs;
100ed062
PN
774 zobj->encoding = REDIS_ENCODING_SKIPLIST;
775 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
a669d5e9
PN
776 unsigned char *zl = ziplistNew();
777
778 if (encoding != REDIS_ENCODING_ZIPLIST)
779 redisPanic("Unknown target encoding");
780
781 /* Approach similar to zslFree(), since we want to free the skiplist at
782 * the same time as creating the ziplist. */
783 zs = zobj->ptr;
784 dictRelease(zs->dict);
785 node = zs->zsl->header->level[0].forward;
786 zfree(zs->zsl->header);
787 zfree(zs->zsl);
788
a669d5e9
PN
789 while (node) {
790 ele = getDecodedObject(node->obj);
8588bfa3 791 zl = zzlInsertAt(zl,NULL,ele,node->score);
a669d5e9
PN
792 decrRefCount(ele);
793
794 next = node->level[0].forward;
795 zslFreeNode(node);
796 node = next;
797 }
798
799 zfree(zs);
8588bfa3 800 zobj->ptr = zl;
a669d5e9
PN
801 zobj->encoding = REDIS_ENCODING_ZIPLIST;
802 } else {
803 redisPanic("Unknown sorted set encoding");
804 }
805}
806
e2641e09 807/*-----------------------------------------------------------------------------
808 * Sorted set commands
809 *----------------------------------------------------------------------------*/
810
69ef89f2 811/* This generic command implements both ZADD and ZINCRBY. */
3ca7532a 812void zaddGenericCommand(redisClient *c, int incr) {
21c5b508 813 static char *nanerr = "resulting score is not a number (NaN)";
3ca7532a
PN
814 robj *key = c->argv[1];
815 robj *ele;
21c5b508
PN
816 robj *zobj;
817 robj *curobj;
ef231a7c 818 double score = 0, *scores, curscore = 0.0;
819 int j, elements = (c->argc-2)/2;
820 int added = 0;
3ca7532a 821
ef231a7c 822 if (c->argc % 2) {
823 addReply(c,shared.syntaxerr);
3ca7532a 824 return;
ef231a7c 825 }
826
827 /* Start parsing all the scores, we need to emit any syntax error
828 * before executing additions to the sorted set, as the command should
829 * either execute fully or nothing at all. */
830 scores = zmalloc(sizeof(double)*elements);
831 for (j = 0; j < elements; j++) {
832 if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
833 != REDIS_OK)
834 {
835 zfree(scores);
836 return;
837 }
838 }
21c5b508 839
ef231a7c 840 /* Lookup the key and create the sorted set if does not exist. */
21c5b508
PN
841 zobj = lookupKeyWrite(c->db,key);
842 if (zobj == NULL) {
a669d5e9
PN
843 if (server.zset_max_ziplist_entries == 0 ||
844 server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
845 {
846 zobj = createZsetObject();
847 } else {
848 zobj = createZsetZiplistObject();
849 }
21c5b508 850 dbAdd(c->db,key,zobj);
e2641e09 851 } else {
21c5b508 852 if (zobj->type != REDIS_ZSET) {
e2641e09 853 addReply(c,shared.wrongtypeerr);
ef231a7c 854 zfree(scores);
e2641e09 855 return;
856 }
857 }
e2641e09 858
ef231a7c 859 for (j = 0; j < elements; j++) {
860 score = scores[j];
861
862 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
863 unsigned char *eptr;
864
865 /* Prefer non-encoded element when dealing with ziplists. */
866 ele = c->argv[3+j*2];
867 if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
868 if (incr) {
869 score += curscore;
870 if (isnan(score)) {
871 addReplyError(c,nanerr);
872 /* Don't need to check if the sorted set is empty
873 * because we know it has at least one element. */
874 zfree(scores);
875 return;
876 }
21c5b508 877 }
21c5b508 878
ef231a7c 879 /* Remove and re-insert when score changed. */
880 if (score != curscore) {
881 zobj->ptr = zzlDelete(zobj->ptr,eptr);
882 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
883
884 signalModifiedKey(c->db,key);
885 server.dirty++;
886 }
887 } else {
888 /* Optimize: check if the element is too large or the list
889 * becomes too long *before* executing zzlInsert. */
8588bfa3 890 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
ef231a7c 891 if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
892 zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
893 if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
894 zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
21c5b508
PN
895
896 signalModifiedKey(c->db,key);
897 server.dirty++;
ef231a7c 898 if (!incr) added++;
21c5b508 899 }
ef231a7c 900 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
901 zset *zs = zobj->ptr;
902 zskiplistNode *znode;
903 dictEntry *de;
904
905 ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
906 de = dictFind(zs->dict,ele);
907 if (de != NULL) {
c0ba9ebe 908 curobj = dictGetKey(de);
909 curscore = *(double*)dictGetVal(de);
ef231a7c 910
911 if (incr) {
912 score += curscore;
913 if (isnan(score)) {
914 addReplyError(c,nanerr);
915 /* Don't need to check if the sorted set is empty
916 * because we know it has at least one element. */
917 zfree(scores);
918 return;
919 }
21c5b508 920 }
21c5b508 921
ef231a7c 922 /* Remove and re-insert when score changed. We can safely
923 * delete the key object from the skiplist, since the
924 * dictionary still has a reference to it. */
925 if (score != curscore) {
eab0e26e 926 redisAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj));
ef231a7c 927 znode = zslInsert(zs->zsl,score,curobj);
928 incrRefCount(curobj); /* Re-inserted in skiplist. */
c0ba9ebe 929 dictGetVal(de) = &znode->score; /* Update score ptr. */
ef231a7c 930
931 signalModifiedKey(c->db,key);
932 server.dirty++;
933 }
934 } else {
935 znode = zslInsert(zs->zsl,score,ele);
936 incrRefCount(ele); /* Inserted in skiplist. */
f013f400 937 redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
ef231a7c 938 incrRefCount(ele); /* Added to dictionary. */
21c5b508
PN
939
940 signalModifiedKey(c->db,key);
941 server.dirty++;
ef231a7c 942 if (!incr) added++;
21c5b508 943 }
21c5b508 944 } else {
ef231a7c 945 redisPanic("Unknown sorted set encoding");
e2641e09 946 }
e2641e09 947 }
ef231a7c 948 zfree(scores);
949 if (incr) /* ZINCRBY */
950 addReplyDouble(c,score);
951 else /* ZADD */
952 addReplyLongLong(c,added);
e2641e09 953}
954
955void zaddCommand(redisClient *c) {
3ca7532a 956 zaddGenericCommand(c,0);
e2641e09 957}
958
959void zincrbyCommand(redisClient *c) {
3ca7532a 960 zaddGenericCommand(c,1);
e2641e09 961}
962
963void zremCommand(redisClient *c) {
0b10e104 964 robj *key = c->argv[1];
0b10e104 965 robj *zobj;
3f7b2b1f 966 int deleted = 0, j;
0b10e104
PN
967
968 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
969 checkType(c,zobj,REDIS_ZSET)) return;
970
971 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
972 unsigned char *eptr;
973
3f7b2b1f 974 for (j = 2; j < c->argc; j++) {
975 if ((eptr = zzlFind(zobj->ptr,c->argv[j],NULL)) != NULL) {
976 deleted++;
977 zobj->ptr = zzlDelete(zobj->ptr,eptr);
978 if (zzlLength(zobj->ptr) == 0) {
979 dbDelete(c->db,key);
980 break;
981 }
982 }
0b10e104 983 }
100ed062 984 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
0b10e104
PN
985 zset *zs = zobj->ptr;
986 dictEntry *de;
987 double score;
988
3f7b2b1f 989 for (j = 2; j < c->argc; j++) {
990 de = dictFind(zs->dict,c->argv[j]);
991 if (de != NULL) {
992 deleted++;
993
994 /* Delete from the skiplist */
c0ba9ebe 995 score = *(double*)dictGetVal(de);
eab0e26e 996 redisAssertWithInfo(c,c->argv[j],zslDelete(zs->zsl,score,c->argv[j]));
3f7b2b1f 997
998 /* Delete from the hash table */
999 dictDelete(zs->dict,c->argv[j]);
1000 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
1001 if (dictSize(zs->dict) == 0) {
1002 dbDelete(c->db,key);
1003 break;
1004 }
1005 }
0b10e104
PN
1006 }
1007 } else {
1008 redisPanic("Unknown sorted set encoding");
e2641e09 1009 }
e2641e09 1010
3f7b2b1f 1011 if (deleted) {
1012 signalModifiedKey(c->db,key);
1013 server.dirty += deleted;
1014 }
1015 addReplyLongLong(c,deleted);
e2641e09 1016}
1017
1018void zremrangebyscoreCommand(redisClient *c) {
4a14dbba
PN
1019 robj *key = c->argv[1];
1020 robj *zobj;
91504b6c 1021 zrangespec range;
4a14dbba 1022 unsigned long deleted;
e2641e09 1023
91504b6c 1024 /* Parse the range arguments. */
7236fdb2 1025 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
d93f9a86 1026 addReplyError(c,"min or max is not a float");
7236fdb2
PN
1027 return;
1028 }
e2641e09 1029
4a14dbba
PN
1030 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
1031 checkType(c,zobj,REDIS_ZSET)) return;
1032
1033 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 1034 zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
48991620 1035 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
100ed062 1036 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
4a14dbba
PN
1037 zset *zs = zobj->ptr;
1038 deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
1039 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
1040 if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
1041 } else {
1042 redisPanic("Unknown sorted set encoding");
1043 }
e2641e09 1044
4a14dbba 1045 if (deleted) signalModifiedKey(c->db,key);
e2641e09 1046 server.dirty += deleted;
1047 addReplyLongLong(c,deleted);
1048}
1049
1050void zremrangebyrankCommand(redisClient *c) {
63b7b7fb
PN
1051 robj *key = c->argv[1];
1052 robj *zobj;
e2641e09 1053 long start;
1054 long end;
1055 int llen;
63b7b7fb 1056 unsigned long deleted;
e2641e09 1057
1058 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
1059 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
1060
63b7b7fb
PN
1061 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
1062 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 1063
63b7b7fb 1064 /* Sanitize indexes. */
df26a0ae 1065 llen = zsetLength(zobj);
e2641e09 1066 if (start < 0) start = llen+start;
1067 if (end < 0) end = llen+end;
1068 if (start < 0) start = 0;
e2641e09 1069
d0a4e24e
PN
1070 /* Invariant: start >= 0, so this test will be true when end < 0.
1071 * The range is empty when start > end or start >= length. */
e2641e09 1072 if (start > end || start >= llen) {
1073 addReply(c,shared.czero);
1074 return;
1075 }
1076 if (end >= llen) end = llen-1;
1077
63b7b7fb
PN
1078 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1079 /* Correct for 1-based rank. */
8588bfa3 1080 zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
48991620 1081 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
100ed062 1082 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
63b7b7fb
PN
1083 zset *zs = zobj->ptr;
1084
1085 /* Correct for 1-based rank. */
1086 deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
1087 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
1088 if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
1089 } else {
1090 redisPanic("Unknown sorted set encoding");
1091 }
1092
1093 if (deleted) signalModifiedKey(c->db,key);
e2641e09 1094 server.dirty += deleted;
63b7b7fb 1095 addReplyLongLong(c,deleted);
e2641e09 1096}
1097
1098typedef struct {
56ce42fa
PN
1099 robj *subject;
1100 int type; /* Set, sorted set */
1101 int encoding;
e2641e09 1102 double weight;
56ce42fa
PN
1103
1104 union {
1105 /* Set iterators. */
1106 union _iterset {
1107 struct {
1108 intset *is;
1109 int ii;
1110 } is;
1111 struct {
1112 dict *dict;
1113 dictIterator *di;
1114 dictEntry *de;
1115 } ht;
1116 } set;
1117
1118 /* Sorted set iterators. */
1119 union _iterzset {
1120 struct {
1121 unsigned char *zl;
1122 unsigned char *eptr, *sptr;
1123 } zl;
1124 struct {
1125 zset *zs;
1126 zskiplistNode *node;
1127 } sl;
1128 } zset;
1129 } iter;
e2641e09 1130} zsetopsrc;
1131
56ce42fa
PN
1132
1133/* Use dirty flags for pointers that need to be cleaned up in the next
1134 * iteration over the zsetopval. The dirty flag for the long long value is
1135 * special, since long long values don't need cleanup. Instead, it means that
1136 * we already checked that "ell" holds a long long, or tried to convert another
1137 * representation into a long long value. When this was successful,
1138 * OPVAL_VALID_LL is set as well. */
1139#define OPVAL_DIRTY_ROBJ 1
1140#define OPVAL_DIRTY_LL 2
1141#define OPVAL_VALID_LL 4
1142
1143/* Store value retrieved from the iterator. */
1144typedef struct {
1145 int flags;
1146 unsigned char _buf[32]; /* Private buffer. */
1147 robj *ele;
1148 unsigned char *estr;
1149 unsigned int elen;
1150 long long ell;
1151 double score;
1152} zsetopval;
1153
1154typedef union _iterset iterset;
1155typedef union _iterzset iterzset;
1156
1157void zuiInitIterator(zsetopsrc *op) {
1158 if (op->subject == NULL)
1159 return;
1160
1161 if (op->type == REDIS_SET) {
1162 iterset *it = &op->iter.set;
1163 if (op->encoding == REDIS_ENCODING_INTSET) {
1164 it->is.is = op->subject->ptr;
1165 it->is.ii = 0;
1166 } else if (op->encoding == REDIS_ENCODING_HT) {
1167 it->ht.dict = op->subject->ptr;
1168 it->ht.di = dictGetIterator(op->subject->ptr);
1169 it->ht.de = dictNext(it->ht.di);
1170 } else {
1171 redisPanic("Unknown set encoding");
1172 }
1173 } else if (op->type == REDIS_ZSET) {
1174 iterzset *it = &op->iter.zset;
1175 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1176 it->zl.zl = op->subject->ptr;
1177 it->zl.eptr = ziplistIndex(it->zl.zl,0);
1178 if (it->zl.eptr != NULL) {
1179 it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
1180 redisAssert(it->zl.sptr != NULL);
1181 }
100ed062 1182 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1183 it->sl.zs = op->subject->ptr;
1184 it->sl.node = it->sl.zs->zsl->header->level[0].forward;
1185 } else {
1186 redisPanic("Unknown sorted set encoding");
1187 }
1188 } else {
1189 redisPanic("Unsupported type");
1190 }
1191}
1192
1193void zuiClearIterator(zsetopsrc *op) {
1194 if (op->subject == NULL)
1195 return;
1196
1197 if (op->type == REDIS_SET) {
1198 iterset *it = &op->iter.set;
1199 if (op->encoding == REDIS_ENCODING_INTSET) {
1200 REDIS_NOTUSED(it); /* skip */
1201 } else if (op->encoding == REDIS_ENCODING_HT) {
1202 dictReleaseIterator(it->ht.di);
1203 } else {
1204 redisPanic("Unknown set encoding");
1205 }
1206 } else if (op->type == REDIS_ZSET) {
1207 iterzset *it = &op->iter.zset;
1208 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1209 REDIS_NOTUSED(it); /* skip */
100ed062 1210 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1211 REDIS_NOTUSED(it); /* skip */
1212 } else {
1213 redisPanic("Unknown sorted set encoding");
1214 }
1215 } else {
1216 redisPanic("Unsupported type");
1217 }
1218}
1219
1220int zuiLength(zsetopsrc *op) {
1221 if (op->subject == NULL)
1222 return 0;
1223
1224 if (op->type == REDIS_SET) {
1225 iterset *it = &op->iter.set;
1226 if (op->encoding == REDIS_ENCODING_INTSET) {
1227 return intsetLen(it->is.is);
1228 } else if (op->encoding == REDIS_ENCODING_HT) {
1229 return dictSize(it->ht.dict);
1230 } else {
1231 redisPanic("Unknown set encoding");
1232 }
1233 } else if (op->type == REDIS_ZSET) {
1234 iterzset *it = &op->iter.zset;
1235 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1236 return zzlLength(it->zl.zl);
100ed062 1237 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1238 return it->sl.zs->zsl->length;
1239 } else {
1240 redisPanic("Unknown sorted set encoding");
1241 }
1242 } else {
1243 redisPanic("Unsupported type");
1244 }
1245}
1246
1247/* Check if the current value is valid. If so, store it in the passed structure
1248 * and move to the next element. If not valid, this means we have reached the
1249 * end of the structure and can abort. */
1250int zuiNext(zsetopsrc *op, zsetopval *val) {
1251 if (op->subject == NULL)
1252 return 0;
1253
1254 if (val->flags & OPVAL_DIRTY_ROBJ)
1255 decrRefCount(val->ele);
1256
fe7be460 1257 memset(val,0,sizeof(zsetopval));
56ce42fa
PN
1258
1259 if (op->type == REDIS_SET) {
1260 iterset *it = &op->iter.set;
1261 if (op->encoding == REDIS_ENCODING_INTSET) {
4dada1b5 1262 int64_t ell;
9de5d460 1263
1264 if (!intsetGet(it->is.is,it->is.ii,&ell))
56ce42fa 1265 return 0;
4dada1b5 1266 val->ell = ell;
56ce42fa
PN
1267 val->score = 1.0;
1268
1269 /* Move to next element. */
1270 it->is.ii++;
1271 } else if (op->encoding == REDIS_ENCODING_HT) {
1272 if (it->ht.de == NULL)
1273 return 0;
c0ba9ebe 1274 val->ele = dictGetKey(it->ht.de);
56ce42fa
PN
1275 val->score = 1.0;
1276
1277 /* Move to next element. */
1278 it->ht.de = dictNext(it->ht.di);
1279 } else {
1280 redisPanic("Unknown set encoding");
1281 }
1282 } else if (op->type == REDIS_ZSET) {
1283 iterzset *it = &op->iter.zset;
1284 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1285 /* No need to check both, but better be explicit. */
1286 if (it->zl.eptr == NULL || it->zl.sptr == NULL)
1287 return 0;
1288 redisAssert(ziplistGet(it->zl.eptr,&val->estr,&val->elen,&val->ell));
1289 val->score = zzlGetScore(it->zl.sptr);
1290
1291 /* Move to next element. */
1292 zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
100ed062 1293 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1294 if (it->sl.node == NULL)
1295 return 0;
1296 val->ele = it->sl.node->obj;
1297 val->score = it->sl.node->score;
1298
1299 /* Move to next element. */
1300 it->sl.node = it->sl.node->level[0].forward;
1301 } else {
1302 redisPanic("Unknown sorted set encoding");
1303 }
1304 } else {
1305 redisPanic("Unsupported type");
1306 }
1307 return 1;
1308}
1309
1310int zuiLongLongFromValue(zsetopval *val) {
1311 if (!(val->flags & OPVAL_DIRTY_LL)) {
1312 val->flags |= OPVAL_DIRTY_LL;
1313
1314 if (val->ele != NULL) {
1315 if (val->ele->encoding == REDIS_ENCODING_INT) {
1316 val->ell = (long)val->ele->ptr;
1317 val->flags |= OPVAL_VALID_LL;
1318 } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
1319 if (string2ll(val->ele->ptr,sdslen(val->ele->ptr),&val->ell))
1320 val->flags |= OPVAL_VALID_LL;
1321 } else {
1322 redisPanic("Unsupported element encoding");
1323 }
1324 } else if (val->estr != NULL) {
1325 if (string2ll((char*)val->estr,val->elen,&val->ell))
1326 val->flags |= OPVAL_VALID_LL;
1327 } else {
1328 /* The long long was already set, flag as valid. */
1329 val->flags |= OPVAL_VALID_LL;
1330 }
1331 }
1332 return val->flags & OPVAL_VALID_LL;
1333}
1334
1335robj *zuiObjectFromValue(zsetopval *val) {
1336 if (val->ele == NULL) {
1337 if (val->estr != NULL) {
1338 val->ele = createStringObject((char*)val->estr,val->elen);
1339 } else {
1340 val->ele = createStringObjectFromLongLong(val->ell);
1341 }
1342 val->flags |= OPVAL_DIRTY_ROBJ;
1343 }
1344 return val->ele;
1345}
1346
1347int zuiBufferFromValue(zsetopval *val) {
1348 if (val->estr == NULL) {
1349 if (val->ele != NULL) {
1350 if (val->ele->encoding == REDIS_ENCODING_INT) {
1351 val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),(long)val->ele->ptr);
1352 val->estr = val->_buf;
1353 } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
1354 val->elen = sdslen(val->ele->ptr);
1355 val->estr = val->ele->ptr;
1356 } else {
1357 redisPanic("Unsupported element encoding");
1358 }
1359 } else {
1360 val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),val->ell);
1361 val->estr = val->_buf;
1362 }
1363 }
1364 return 1;
1365}
1366
1367/* Find value pointed to by val in the source pointer to by op. When found,
1368 * return 1 and store its score in target. Return 0 otherwise. */
1369int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
1370 if (op->subject == NULL)
1371 return 0;
1372
1373 if (op->type == REDIS_SET) {
1374 iterset *it = &op->iter.set;
1375
1376 if (op->encoding == REDIS_ENCODING_INTSET) {
1377 if (zuiLongLongFromValue(val) && intsetFind(it->is.is,val->ell)) {
1378 *score = 1.0;
1379 return 1;
1380 } else {
1381 return 0;
1382 }
1383 } else if (op->encoding == REDIS_ENCODING_HT) {
1384 zuiObjectFromValue(val);
1385 if (dictFind(it->ht.dict,val->ele) != NULL) {
1386 *score = 1.0;
1387 return 1;
1388 } else {
1389 return 0;
1390 }
1391 } else {
1392 redisPanic("Unknown set encoding");
1393 }
1394 } else if (op->type == REDIS_ZSET) {
1395 iterzset *it = &op->iter.zset;
1396 zuiObjectFromValue(val);
1397
1398 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 1399 if (zzlFind(it->zl.zl,val->ele,score) != NULL) {
56ce42fa
PN
1400 /* Score is already set by zzlFind. */
1401 return 1;
1402 } else {
1403 return 0;
1404 }
100ed062 1405 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1406 dictEntry *de;
1407 if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
c0ba9ebe 1408 *score = *(double*)dictGetVal(de);
56ce42fa
PN
1409 return 1;
1410 } else {
1411 return 0;
1412 }
1413 } else {
1414 redisPanic("Unknown sorted set encoding");
1415 }
1416 } else {
1417 redisPanic("Unsupported type");
1418 }
1419}
1420
1421int zuiCompareByCardinality(const void *s1, const void *s2) {
1422 return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2);
e2641e09 1423}
1424
1425#define REDIS_AGGR_SUM 1
1426#define REDIS_AGGR_MIN 2
1427#define REDIS_AGGR_MAX 3
c0ba9ebe 1428#define zunionInterDictValue(_e) (dictGetVal(_e) == NULL ? 1.0 : *(double*)dictGetVal(_e))
e2641e09 1429
1430inline static void zunionInterAggregate(double *target, double val, int aggregate) {
1431 if (aggregate == REDIS_AGGR_SUM) {
1432 *target = *target + val;
d9e28bcf
PN
1433 /* The result of adding two doubles is NaN when one variable
1434 * is +inf and the other is -inf. When these numbers are added,
1435 * we maintain the convention of the result being 0.0. */
1436 if (isnan(*target)) *target = 0.0;
e2641e09 1437 } else if (aggregate == REDIS_AGGR_MIN) {
1438 *target = val < *target ? val : *target;
1439 } else if (aggregate == REDIS_AGGR_MAX) {
1440 *target = val > *target ? val : *target;
1441 } else {
1442 /* safety net */
1443 redisPanic("Unknown ZUNION/INTER aggregate type");
1444 }
1445}
1446
1447void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
706b32e0
B
1448 int i, j;
1449 long setnum;
e2641e09 1450 int aggregate = REDIS_AGGR_SUM;
1451 zsetopsrc *src;
56ce42fa
PN
1452 zsetopval zval;
1453 robj *tmp;
255eebe2 1454 unsigned int maxelelen = 0;
e2641e09 1455 robj *dstobj;
1456 zset *dstzset;
69ef89f2 1457 zskiplistNode *znode;
8c1420ff 1458 int touched = 0;
e2641e09 1459
1460 /* expect setnum input keys to be given */
706b32e0
B
1461 if ((getLongFromObjectOrReply(c, c->argv[2], &setnum, NULL) != REDIS_OK))
1462 return;
1463
e2641e09 1464 if (setnum < 1) {
3ab20376
PN
1465 addReplyError(c,
1466 "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE");
e2641e09 1467 return;
1468 }
1469
1470 /* test if the expected number of keys would overflow */
1471 if (3+setnum > c->argc) {
1472 addReply(c,shared.syntaxerr);
1473 return;
1474 }
1475
1476 /* read keys to be used for input */
56ce42fa 1477 src = zcalloc(sizeof(zsetopsrc) * setnum);
e2641e09 1478 for (i = 0, j = 3; i < setnum; i++, j++) {
1479 robj *obj = lookupKeyWrite(c->db,c->argv[j]);
56ce42fa
PN
1480 if (obj != NULL) {
1481 if (obj->type != REDIS_ZSET && obj->type != REDIS_SET) {
e2641e09 1482 zfree(src);
1483 addReply(c,shared.wrongtypeerr);
1484 return;
1485 }
56ce42fa
PN
1486
1487 src[i].subject = obj;
1488 src[i].type = obj->type;
1489 src[i].encoding = obj->encoding;
1490 } else {
1491 src[i].subject = NULL;
e2641e09 1492 }
1493
56ce42fa 1494 /* Default all weights to 1. */
e2641e09 1495 src[i].weight = 1.0;
1496 }
1497
1498 /* parse optional extra arguments */
1499 if (j < c->argc) {
1500 int remaining = c->argc - j;
1501
1502 while (remaining) {
1503 if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
1504 j++; remaining--;
1505 for (i = 0; i < setnum; i++, j++, remaining--) {
673e1fb7 1506 if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
d93f9a86 1507 "weight value is not a float") != REDIS_OK)
673e1fb7
PN
1508 {
1509 zfree(src);
e2641e09 1510 return;
673e1fb7 1511 }
e2641e09 1512 }
1513 } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
1514 j++; remaining--;
1515 if (!strcasecmp(c->argv[j]->ptr,"sum")) {
1516 aggregate = REDIS_AGGR_SUM;
1517 } else if (!strcasecmp(c->argv[j]->ptr,"min")) {
1518 aggregate = REDIS_AGGR_MIN;
1519 } else if (!strcasecmp(c->argv[j]->ptr,"max")) {
1520 aggregate = REDIS_AGGR_MAX;
1521 } else {
1522 zfree(src);
1523 addReply(c,shared.syntaxerr);
1524 return;
1525 }
1526 j++; remaining--;
1527 } else {
1528 zfree(src);
1529 addReply(c,shared.syntaxerr);
1530 return;
1531 }
1532 }
1533 }
1534
56ce42fa
PN
1535 for (i = 0; i < setnum; i++)
1536 zuiInitIterator(&src[i]);
1537
e2641e09 1538 /* sort sets from the smallest to largest, this will improve our
1539 * algorithm's performance */
56ce42fa 1540 qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
e2641e09 1541
1542 dstobj = createZsetObject();
1543 dstzset = dstobj->ptr;
02e60065 1544 memset(&zval, 0, sizeof(zval));
e2641e09 1545
1546 if (op == REDIS_OP_INTER) {
56ce42fa
PN
1547 /* Skip everything if the smallest input is empty. */
1548 if (zuiLength(&src[0]) > 0) {
1549 /* Precondition: as src[0] is non-empty and the inputs are ordered
1550 * by size, all src[i > 0] are non-empty too. */
1551 while (zuiNext(&src[0],&zval)) {
d433ebc6 1552 double score, value;
e2641e09 1553
56ce42fa 1554 score = src[0].weight * zval.score;
256356ff 1555 if (isnan(score)) score = 0;
1556
e2641e09 1557 for (j = 1; j < setnum; j++) {
d200342a 1558 /* It is not safe to access the zset we are
cb16b6c3 1559 * iterating, so explicitly check for equal object. */
d070abe4 1560 if (src[j].subject == src[0].subject) {
1561 value = zval.score*src[j].weight;
1562 zunionInterAggregate(&score,value,aggregate);
1563 } else if (zuiFind(&src[j],&zval,&value)) {
56ce42fa 1564 value *= src[j].weight;
d433ebc6 1565 zunionInterAggregate(&score,value,aggregate);
e2641e09 1566 } else {
1567 break;
1568 }
1569 }
1570
56ce42fa 1571 /* Only continue when present in every input. */
d433ebc6 1572 if (j == setnum) {
56ce42fa
PN
1573 tmp = zuiObjectFromValue(&zval);
1574 znode = zslInsert(dstzset->zsl,score,tmp);
1575 incrRefCount(tmp); /* added to skiplist */
1576 dictAdd(dstzset->dict,tmp,&znode->score);
1577 incrRefCount(tmp); /* added to dictionary */
255eebe2
PN
1578
1579 if (tmp->encoding == REDIS_ENCODING_RAW)
1580 if (sdslen(tmp->ptr) > maxelelen)
1581 maxelelen = sdslen(tmp->ptr);
e2641e09 1582 }
1583 }
e2641e09 1584 }
1585 } else if (op == REDIS_OP_UNION) {
1586 for (i = 0; i < setnum; i++) {
521ddcce 1587 if (zuiLength(&src[i]) == 0)
56ce42fa 1588 continue;
e2641e09 1589
56ce42fa 1590 while (zuiNext(&src[i],&zval)) {
d433ebc6
PN
1591 double score, value;
1592
56ce42fa
PN
1593 /* Skip key when already processed */
1594 if (dictFind(dstzset->dict,zuiObjectFromValue(&zval)) != NULL)
d433ebc6 1595 continue;
e2641e09 1596
56ce42fa
PN
1597 /* Initialize score */
1598 score = src[i].weight * zval.score;
256356ff 1599 if (isnan(score)) score = 0;
e2641e09 1600
56ce42fa
PN
1601 /* Because the inputs are sorted by size, it's only possible
1602 * for sets at larger indices to hold this element. */
e2641e09 1603 for (j = (i+1); j < setnum; j++) {
d200342a 1604 /* It is not safe to access the zset we are
cb16b6c3 1605 * iterating, so explicitly check for equal object. */
1606 if(src[j].subject == src[i].subject) {
1607 value = zval.score*src[j].weight;
1608 zunionInterAggregate(&score,value,aggregate);
1609 } else if (zuiFind(&src[j],&zval,&value)) {
56ce42fa 1610 value *= src[j].weight;
d433ebc6 1611 zunionInterAggregate(&score,value,aggregate);
e2641e09 1612 }
1613 }
1614
56ce42fa
PN
1615 tmp = zuiObjectFromValue(&zval);
1616 znode = zslInsert(dstzset->zsl,score,tmp);
1617 incrRefCount(zval.ele); /* added to skiplist */
1618 dictAdd(dstzset->dict,tmp,&znode->score);
1619 incrRefCount(zval.ele); /* added to dictionary */
255eebe2
PN
1620
1621 if (tmp->encoding == REDIS_ENCODING_RAW)
1622 if (sdslen(tmp->ptr) > maxelelen)
1623 maxelelen = sdslen(tmp->ptr);
e2641e09 1624 }
e2641e09 1625 }
1626 } else {
56ce42fa 1627 redisPanic("Unknown operator");
e2641e09 1628 }
1629
56ce42fa
PN
1630 for (i = 0; i < setnum; i++)
1631 zuiClearIterator(&src[i]);
1632
8c1420ff 1633 if (dbDelete(c->db,dstkey)) {
cea8c5cd 1634 signalModifiedKey(c->db,dstkey);
8c1420ff 1635 touched = 1;
1636 server.dirty++;
1637 }
e2641e09 1638 if (dstzset->zsl->length) {
255eebe2
PN
1639 /* Convert to ziplist when in limits. */
1640 if (dstzset->zsl->length <= server.zset_max_ziplist_entries &&
1641 maxelelen <= server.zset_max_ziplist_value)
df26a0ae 1642 zsetConvert(dstobj,REDIS_ENCODING_ZIPLIST);
255eebe2 1643
e2641e09 1644 dbAdd(c->db,dstkey,dstobj);
df26a0ae 1645 addReplyLongLong(c,zsetLength(dstobj));
cea8c5cd 1646 if (!touched) signalModifiedKey(c->db,dstkey);
cbf7e107 1647 server.dirty++;
e2641e09 1648 } else {
1649 decrRefCount(dstobj);
255eebe2 1650 addReply(c,shared.czero);
e2641e09 1651 }
1652 zfree(src);
1653}
1654
1655void zunionstoreCommand(redisClient *c) {
1656 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_UNION);
1657}
1658
1659void zinterstoreCommand(redisClient *c) {
1660 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_INTER);
1661}
1662
1663void zrangeGenericCommand(redisClient *c, int reverse) {
5d1b4fb6
PN
1664 robj *key = c->argv[1];
1665 robj *zobj;
1666 int withscores = 0;
e2641e09 1667 long start;
1668 long end;
e2641e09 1669 int llen;
5d1b4fb6 1670 int rangelen;
e2641e09 1671
1672 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
1673 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
1674
1675 if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
1676 withscores = 1;
1677 } else if (c->argc >= 5) {
1678 addReply(c,shared.syntaxerr);
1679 return;
1680 }
1681
5d1b4fb6
PN
1682 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL
1683 || checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 1684
5d1b4fb6 1685 /* Sanitize indexes. */
df26a0ae 1686 llen = zsetLength(zobj);
e2641e09 1687 if (start < 0) start = llen+start;
1688 if (end < 0) end = llen+end;
1689 if (start < 0) start = 0;
e2641e09 1690
d0a4e24e
PN
1691 /* Invariant: start >= 0, so this test will be true when end < 0.
1692 * The range is empty when start > end or start >= length. */
e2641e09 1693 if (start > end || start >= llen) {
e2641e09 1694 addReply(c,shared.emptymultibulk);
1695 return;
1696 }
1697 if (end >= llen) end = llen-1;
1698 rangelen = (end-start)+1;
1699
e2641e09 1700 /* Return the result in form of a multi-bulk reply */
5d1b4fb6
PN
1701 addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen);
1702
1703 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1704 unsigned char *zl = zobj->ptr;
1705 unsigned char *eptr, *sptr;
1706 unsigned char *vstr;
1707 unsigned int vlen;
1708 long long vlong;
1709
1710 if (reverse)
1711 eptr = ziplistIndex(zl,-2-(2*start));
1712 else
1713 eptr = ziplistIndex(zl,2*start);
1714
eab0e26e 1715 redisAssertWithInfo(c,zobj,eptr != NULL);
4c5f0966
PN
1716 sptr = ziplistNext(zl,eptr);
1717
5d1b4fb6 1718 while (rangelen--) {
eab0e26e 1719 redisAssertWithInfo(c,zobj,eptr != NULL && sptr != NULL);
1720 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
5d1b4fb6
PN
1721 if (vstr == NULL)
1722 addReplyBulkLongLong(c,vlong);
1723 else
1724 addReplyBulkCBuffer(c,vstr,vlen);
1725
4c5f0966 1726 if (withscores)
5d1b4fb6 1727 addReplyDouble(c,zzlGetScore(sptr));
5d1b4fb6 1728
4c5f0966
PN
1729 if (reverse)
1730 zzlPrev(zl,&eptr,&sptr);
1731 else
1732 zzlNext(zl,&eptr,&sptr);
5d1b4fb6
PN
1733 }
1734
100ed062 1735 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
5d1b4fb6
PN
1736 zset *zs = zobj->ptr;
1737 zskiplist *zsl = zs->zsl;
1738 zskiplistNode *ln;
1739 robj *ele;
1740
1741 /* Check if starting point is trivial, before doing log(N) lookup. */
1742 if (reverse) {
1743 ln = zsl->tail;
1744 if (start > 0)
1745 ln = zslGetElementByRank(zsl,llen-start);
1746 } else {
1747 ln = zsl->header->level[0].forward;
1748 if (start > 0)
1749 ln = zslGetElementByRank(zsl,start+1);
1750 }
1751
1752 while(rangelen--) {
eab0e26e 1753 redisAssertWithInfo(c,zobj,ln != NULL);
5d1b4fb6
PN
1754 ele = ln->obj;
1755 addReplyBulk(c,ele);
1756 if (withscores)
1757 addReplyDouble(c,ln->score);
1758 ln = reverse ? ln->backward : ln->level[0].forward;
1759 }
1760 } else {
1761 redisPanic("Unknown sorted set encoding");
e2641e09 1762 }
1763}
1764
1765void zrangeCommand(redisClient *c) {
1766 zrangeGenericCommand(c,0);
1767}
1768
1769void zrevrangeCommand(redisClient *c) {
1770 zrangeGenericCommand(c,1);
1771}
1772
0cfc8940
PN
1773/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
1774void genericZrangebyscoreCommand(redisClient *c, int reverse) {
25bb8a44 1775 zrangespec range;
aff255c8 1776 robj *key = c->argv[1];
0cfc8940 1777 robj *zobj;
706b32e0 1778 long offset = 0, limit = -1;
e2641e09 1779 int withscores = 0;
25bb8a44
PN
1780 unsigned long rangelen = 0;
1781 void *replylen = NULL;
22b9bf15 1782 int minidx, maxidx;
e2641e09 1783
25bb8a44 1784 /* Parse the range arguments. */
22b9bf15
PN
1785 if (reverse) {
1786 /* Range is given as [max,min] */
1787 maxidx = 2; minidx = 3;
1788 } else {
1789 /* Range is given as [min,max] */
1790 minidx = 2; maxidx = 3;
1791 }
1792
1793 if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != REDIS_OK) {
d93f9a86 1794 addReplyError(c,"min or max is not a float");
7236fdb2
PN
1795 return;
1796 }
25bb8a44
PN
1797
1798 /* Parse optional extra arguments. Note that ZCOUNT will exactly have
1799 * 4 arguments, so we'll never enter the following code path. */
1800 if (c->argc > 4) {
1801 int remaining = c->argc - 4;
1802 int pos = 4;
1803
1804 while (remaining) {
1805 if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) {
1806 pos++; remaining--;
1807 withscores = 1;
1808 } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {
706b32e0
B
1809 if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != REDIS_OK) ||
1810 (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != REDIS_OK)) return;
25bb8a44
PN
1811 pos += 3; remaining -= 3;
1812 } else {
1813 addReply(c,shared.syntaxerr);
1814 return;
1815 }
1816 }
e2641e09 1817 }
25bb8a44
PN
1818
1819 /* Ok, lookup the key and get the range */
0cfc8940 1820 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL ||
aff255c8 1821 checkType(c,zobj,REDIS_ZSET)) return;
25bb8a44 1822
aff255c8
PN
1823 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1824 unsigned char *zl = zobj->ptr;
1825 unsigned char *eptr, *sptr;
1826 unsigned char *vstr;
1827 unsigned int vlen;
1828 long long vlong;
1829 double score;
e2641e09 1830
aff255c8 1831 /* If reversed, get the last node in range as starting point. */
0cfc8940 1832 if (reverse) {
8588bfa3 1833 eptr = zzlLastInRange(zl,range);
0cfc8940 1834 } else {
8588bfa3 1835 eptr = zzlFirstInRange(zl,range);
0cfc8940 1836 }
e2641e09 1837
aff255c8
PN
1838 /* No "first" element in the specified interval. */
1839 if (eptr == NULL) {
0cfc8940 1840 addReply(c, shared.emptymultibulk);
aff255c8
PN
1841 return;
1842 }
e2641e09 1843
aff255c8 1844 /* Get score pointer for the first element. */
eab0e26e 1845 redisAssertWithInfo(c,zobj,eptr != NULL);
aff255c8 1846 sptr = ziplistNext(zl,eptr);
e2641e09 1847
aff255c8
PN
1848 /* We don't know in advance how many matching elements there are in the
1849 * list, so we push this object that will represent the multi-bulk
1850 * length in the output buffer, and will "fix" it later */
0cfc8940 1851 replylen = addDeferredMultiBulkLength(c);
aff255c8
PN
1852
1853 /* If there is an offset, just traverse the number of elements without
1854 * checking the score because that is done in the next loop. */
0cfc8940
PN
1855 while (eptr && offset--) {
1856 if (reverse) {
aff255c8 1857 zzlPrev(zl,&eptr,&sptr);
0cfc8940 1858 } else {
aff255c8 1859 zzlNext(zl,&eptr,&sptr);
0cfc8940
PN
1860 }
1861 }
aff255c8
PN
1862
1863 while (eptr && limit--) {
1864 score = zzlGetScore(sptr);
1865
1866 /* Abort when the node is no longer in range. */
1867 if (reverse) {
1868 if (!zslValueGteMin(score,&range)) break;
1869 } else {
1870 if (!zslValueLteMax(score,&range)) break;
1871 }
1872
0cfc8940 1873 /* We know the element exists, so ziplistGet should always succeed */
eab0e26e 1874 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
0cfc8940 1875
aff255c8 1876 rangelen++;
0cfc8940
PN
1877 if (vstr == NULL) {
1878 addReplyBulkLongLong(c,vlong);
1879 } else {
1880 addReplyBulkCBuffer(c,vstr,vlen);
1881 }
1882
1883 if (withscores) {
1884 addReplyDouble(c,score);
aff255c8
PN
1885 }
1886
1887 /* Move to next node */
0cfc8940 1888 if (reverse) {
aff255c8 1889 zzlPrev(zl,&eptr,&sptr);
0cfc8940 1890 } else {
aff255c8 1891 zzlNext(zl,&eptr,&sptr);
0cfc8940 1892 }
e2641e09 1893 }
100ed062 1894 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
aff255c8
PN
1895 zset *zs = zobj->ptr;
1896 zskiplist *zsl = zs->zsl;
1897 zskiplistNode *ln;
25bb8a44 1898
aff255c8 1899 /* If reversed, get the last node in range as starting point. */
0cfc8940 1900 if (reverse) {
aff255c8 1901 ln = zslLastInRange(zsl,range);
0cfc8940 1902 } else {
aff255c8 1903 ln = zslFirstInRange(zsl,range);
0cfc8940 1904 }
aff255c8
PN
1905
1906 /* No "first" element in the specified interval. */
1907 if (ln == NULL) {
0cfc8940 1908 addReply(c, shared.emptymultibulk);
aff255c8 1909 return;
25bb8a44
PN
1910 }
1911
aff255c8
PN
1912 /* We don't know in advance how many matching elements there are in the
1913 * list, so we push this object that will represent the multi-bulk
1914 * length in the output buffer, and will "fix" it later */
0cfc8940 1915 replylen = addDeferredMultiBulkLength(c);
aff255c8
PN
1916
1917 /* If there is an offset, just traverse the number of elements without
1918 * checking the score because that is done in the next loop. */
0cfc8940
PN
1919 while (ln && offset--) {
1920 if (reverse) {
1921 ln = ln->backward;
1922 } else {
1923 ln = ln->level[0].forward;
1924 }
1925 }
aff255c8
PN
1926
1927 while (ln && limit--) {
1928 /* Abort when the node is no longer in range. */
1929 if (reverse) {
1930 if (!zslValueGteMin(ln->score,&range)) break;
1931 } else {
1932 if (!zslValueLteMax(ln->score,&range)) break;
1933 }
1934
aff255c8 1935 rangelen++;
0cfc8940
PN
1936 addReplyBulk(c,ln->obj);
1937
1938 if (withscores) {
1939 addReplyDouble(c,ln->score);
aff255c8
PN
1940 }
1941
1942 /* Move to next node */
0cfc8940
PN
1943 if (reverse) {
1944 ln = ln->backward;
1945 } else {
1946 ln = ln->level[0].forward;
1947 }
aff255c8
PN
1948 }
1949 } else {
1950 redisPanic("Unknown sorted set encoding");
25bb8a44
PN
1951 }
1952
0cfc8940
PN
1953 if (withscores) {
1954 rangelen *= 2;
e2641e09 1955 }
0cfc8940
PN
1956
1957 setDeferredMultiBulkLength(c, replylen, rangelen);
e2641e09 1958}
1959
1960void zrangebyscoreCommand(redisClient *c) {
0cfc8940 1961 genericZrangebyscoreCommand(c,0);
25bb8a44
PN
1962}
1963
1964void zrevrangebyscoreCommand(redisClient *c) {
0cfc8940 1965 genericZrangebyscoreCommand(c,1);
e2641e09 1966}
1967
1968void zcountCommand(redisClient *c) {
62d774e5
PN
1969 robj *key = c->argv[1];
1970 robj *zobj;
1971 zrangespec range;
1972 int count = 0;
1973
1974 /* Parse the range arguments */
1975 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
d93f9a86 1976 addReplyError(c,"min or max is not a float");
62d774e5
PN
1977 return;
1978 }
1979
1980 /* Lookup the sorted set */
1981 if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||
1982 checkType(c, zobj, REDIS_ZSET)) return;
1983
1984 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1985 unsigned char *zl = zobj->ptr;
1986 unsigned char *eptr, *sptr;
1987 double score;
1988
1989 /* Use the first element in range as the starting point */
1990 eptr = zzlFirstInRange(zl,range);
1991
1992 /* No "first" element */
1993 if (eptr == NULL) {
1994 addReply(c, shared.czero);
1995 return;
1996 }
1997
1998 /* First element is in range */
1999 sptr = ziplistNext(zl,eptr);
2000 score = zzlGetScore(sptr);
eab0e26e 2001 redisAssertWithInfo(c,zobj,zslValueLteMax(score,&range));
62d774e5
PN
2002
2003 /* Iterate over elements in range */
2004 while (eptr) {
2005 score = zzlGetScore(sptr);
2006
2007 /* Abort when the node is no longer in range. */
2008 if (!zslValueLteMax(score,&range)) {
2009 break;
2010 } else {
2011 count++;
2012 zzlNext(zl,&eptr,&sptr);
2013 }
2014 }
2015 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
2016 zset *zs = zobj->ptr;
2017 zskiplist *zsl = zs->zsl;
2018 zskiplistNode *zn;
2019 unsigned long rank;
2020
2021 /* Find first element in range */
2022 zn = zslFirstInRange(zsl, range);
2023
2024 /* Use rank of first element, if any, to determine preliminary count */
2025 if (zn != NULL) {
2026 rank = zslGetRank(zsl, zn->score, zn->obj);
2027 count = (zsl->length - (rank - 1));
2028
2029 /* Find last element in range */
2030 zn = zslLastInRange(zsl, range);
2031
2032 /* Use rank of last element, if any, to determine the actual count */
2033 if (zn != NULL) {
2034 rank = zslGetRank(zsl, zn->score, zn->obj);
2035 count -= (zsl->length - rank);
2036 }
2037 }
2038 } else {
2039 redisPanic("Unknown sorted set encoding");
2040 }
2041
2042 addReplyLongLong(c, count);
e2641e09 2043}
2044
2045void zcardCommand(redisClient *c) {
d1c920c5
PN
2046 robj *key = c->argv[1];
2047 robj *zobj;
e2641e09 2048
d1c920c5
PN
2049 if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||
2050 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 2051
df26a0ae 2052 addReplyLongLong(c,zsetLength(zobj));
e2641e09 2053}
2054
2055void zscoreCommand(redisClient *c) {
d1c920c5
PN
2056 robj *key = c->argv[1];
2057 robj *zobj;
2058 double score;
e2641e09 2059
d1c920c5
PN
2060 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
2061 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 2062
d1c920c5 2063 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 2064 if (zzlFind(zobj->ptr,c->argv[2],&score) != NULL)
d1c920c5
PN
2065 addReplyDouble(c,score);
2066 else
2067 addReply(c,shared.nullbulk);
100ed062 2068 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
d1c920c5
PN
2069 zset *zs = zobj->ptr;
2070 dictEntry *de;
e2641e09 2071
d1c920c5
PN
2072 c->argv[2] = tryObjectEncoding(c->argv[2]);
2073 de = dictFind(zs->dict,c->argv[2]);
2074 if (de != NULL) {
c0ba9ebe 2075 score = *(double*)dictGetVal(de);
d1c920c5
PN
2076 addReplyDouble(c,score);
2077 } else {
2078 addReply(c,shared.nullbulk);
2079 }
2080 } else {
2081 redisPanic("Unknown sorted set encoding");
e2641e09 2082 }
2083}
2084
2085void zrankGenericCommand(redisClient *c, int reverse) {
d1c920c5
PN
2086 robj *key = c->argv[1];
2087 robj *ele = c->argv[2];
2088 robj *zobj;
2089 unsigned long llen;
e2641e09 2090 unsigned long rank;
e2641e09 2091
d1c920c5
PN
2092 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
2093 checkType(c,zobj,REDIS_ZSET)) return;
df26a0ae 2094 llen = zsetLength(zobj);
e2641e09 2095
eab0e26e 2096 redisAssertWithInfo(c,ele,ele->encoding == REDIS_ENCODING_RAW);
d1c920c5
PN
2097 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
2098 unsigned char *zl = zobj->ptr;
2099 unsigned char *eptr, *sptr;
e2641e09 2100
d1c920c5 2101 eptr = ziplistIndex(zl,0);
eab0e26e 2102 redisAssertWithInfo(c,zobj,eptr != NULL);
d1c920c5 2103 sptr = ziplistNext(zl,eptr);
eab0e26e 2104 redisAssertWithInfo(c,zobj,sptr != NULL);
d1c920c5
PN
2105
2106 rank = 1;
2107 while(eptr != NULL) {
2108 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr)))
2109 break;
2110 rank++;
2111 zzlNext(zl,&eptr,&sptr);
2112 }
2113
2114 if (eptr != NULL) {
2115 if (reverse)
2116 addReplyLongLong(c,llen-rank);
2117 else
2118 addReplyLongLong(c,rank-1);
e2641e09 2119 } else {
d1c920c5
PN
2120 addReply(c,shared.nullbulk);
2121 }
100ed062 2122 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
d1c920c5
PN
2123 zset *zs = zobj->ptr;
2124 zskiplist *zsl = zs->zsl;
2125 dictEntry *de;
2126 double score;
2127
2128 ele = c->argv[2] = tryObjectEncoding(c->argv[2]);
2129 de = dictFind(zs->dict,ele);
2130 if (de != NULL) {
c0ba9ebe 2131 score = *(double*)dictGetVal(de);
d1c920c5 2132 rank = zslGetRank(zsl,score,ele);
eab0e26e 2133 redisAssertWithInfo(c,ele,rank); /* Existing elements always have a rank. */
d1c920c5
PN
2134 if (reverse)
2135 addReplyLongLong(c,llen-rank);
2136 else
2137 addReplyLongLong(c,rank-1);
2138 } else {
2139 addReply(c,shared.nullbulk);
e2641e09 2140 }
2141 } else {
d1c920c5 2142 redisPanic("Unknown sorted set encoding");
e2641e09 2143 }
2144}
2145
2146void zrankCommand(redisClient *c) {
2147 zrankGenericCommand(c, 0);
2148}
2149
2150void zrevrankCommand(redisClient *c) {
2151 zrankGenericCommand(c, 1);
2152}