]> git.saurik.com Git - redis.git/blob - src/t_zset.c
e528205a73ce1ff515d2985e821eca49f25e7983
[redis.git] / src / t_zset.c
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:
20 * a) this implementation allows for repeated values.
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
26 zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
27 zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
28 zn->score = score;
29 zn->obj = obj;
30 return zn;
31 }
32
33 zskiplist *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++) {
42 zsl->header->level[j].forward = NULL;
43 zsl->header->level[j].span = 0;
44 }
45 zsl->header->backward = NULL;
46 zsl->tail = NULL;
47 return zsl;
48 }
49
50 void zslFreeNode(zskiplistNode *node) {
51 decrRefCount(node->obj);
52 zfree(node);
53 }
54
55 void zslFree(zskiplist *zsl) {
56 zskiplistNode *node = zsl->header->level[0].forward, *next;
57
58 zfree(zsl->header);
59 while(node) {
60 next = node->level[0].forward;
61 zslFreeNode(node);
62 node = next;
63 }
64 zfree(zsl);
65 }
66
67 int zslRandomLevel(void) {
68 int level = 1;
69 while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
70 level += 1;
71 return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
72 }
73
74 zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
75 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
76 unsigned int rank[ZSKIPLIST_MAXLEVEL];
77 int i, level;
78
79 x = zsl->header;
80 for (i = zsl->level-1; i >= 0; i--) {
81 /* store rank that is crossed to reach the insert position */
82 rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
83 while (x->level[i].forward &&
84 (x->level[i].forward->score < score ||
85 (x->level[i].forward->score == score &&
86 compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
87 rank[i] += x->level[i].span;
88 x = x->level[i].forward;
89 }
90 update[i] = x;
91 }
92 /* we assume the key is not already inside, since we allow duplicated
93 * scores, and the re-insertion of score and redis object should never
94 * happpen since the caller of zslInsert() should test in the hash table
95 * if the element is already inside or not. */
96 level = zslRandomLevel();
97 if (level > zsl->level) {
98 for (i = zsl->level; i < level; i++) {
99 rank[i] = 0;
100 update[i] = zsl->header;
101 update[i]->level[i].span = zsl->length;
102 }
103 zsl->level = level;
104 }
105 x = zslCreateNode(level,score,obj);
106 for (i = 0; i < level; i++) {
107 x->level[i].forward = update[i]->level[i].forward;
108 update[i]->level[i].forward = x;
109
110 /* update span covered by update[i] as x is inserted here */
111 x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
112 update[i]->level[i].span = (rank[0] - rank[i]) + 1;
113 }
114
115 /* increment span for untouched levels */
116 for (i = level; i < zsl->level; i++) {
117 update[i]->level[i].span++;
118 }
119
120 x->backward = (update[0] == zsl->header) ? NULL : update[0];
121 if (x->level[0].forward)
122 x->level[0].forward->backward = x;
123 else
124 zsl->tail = x;
125 zsl->length++;
126 return x;
127 }
128
129 /* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
130 void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
131 int i;
132 for (i = 0; i < zsl->level; i++) {
133 if (update[i]->level[i].forward == x) {
134 update[i]->level[i].span += x->level[i].span - 1;
135 update[i]->level[i].forward = x->level[i].forward;
136 } else {
137 update[i]->level[i].span -= 1;
138 }
139 }
140 if (x->level[0].forward) {
141 x->level[0].forward->backward = x->backward;
142 } else {
143 zsl->tail = x->backward;
144 }
145 while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
146 zsl->level--;
147 zsl->length--;
148 }
149
150 /* Delete an element with matching score/object from the skiplist. */
151 int zslDelete(zskiplist *zsl, double score, robj *obj) {
152 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
153 int i;
154
155 x = zsl->header;
156 for (i = zsl->level-1; i >= 0; i--) {
157 while (x->level[i].forward &&
158 (x->level[i].forward->score < score ||
159 (x->level[i].forward->score == score &&
160 compareStringObjects(x->level[i].forward->obj,obj) < 0)))
161 x = x->level[i].forward;
162 update[i] = x;
163 }
164 /* We may have multiple elements with the same score, what we need
165 * is to find the element with both the right score and object. */
166 x = x->level[0].forward;
167 if (x && score == x->score && equalStringObjects(x->obj,obj)) {
168 zslDeleteNode(zsl, x, update);
169 zslFreeNode(x);
170 return 1;
171 } else {
172 return 0; /* not found */
173 }
174 return 0; /* not found */
175 }
176
177 /* Delete all the elements with score between min and max from the skiplist.
178 * Min and mx are inclusive, so a score >= min || score <= max is deleted.
179 * Note that this function takes the reference to the hash table view of the
180 * sorted set, in order to remove the elements from the hash table too. */
181 unsigned long zslDeleteRangeByScore(zskiplist *zsl, double min, double max, dict *dict) {
182 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
183 unsigned long removed = 0;
184 int i;
185
186 x = zsl->header;
187 for (i = zsl->level-1; i >= 0; i--) {
188 while (x->level[i].forward && x->level[i].forward->score < min)
189 x = x->level[i].forward;
190 update[i] = x;
191 }
192 /* We may have multiple elements with the same score, what we need
193 * is to find the element with both the right score and object. */
194 x = x->level[0].forward;
195 while (x && x->score <= max) {
196 zskiplistNode *next = x->level[0].forward;
197 zslDeleteNode(zsl,x,update);
198 dictDelete(dict,x->obj);
199 zslFreeNode(x);
200 removed++;
201 x = next;
202 }
203 return removed; /* not found */
204 }
205
206 /* Delete all the elements with rank between start and end from the skiplist.
207 * Start and end are inclusive. Note that start and end need to be 1-based */
208 unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
209 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
210 unsigned long traversed = 0, removed = 0;
211 int i;
212
213 x = zsl->header;
214 for (i = zsl->level-1; i >= 0; i--) {
215 while (x->level[i].forward && (traversed + x->level[i].span) < start) {
216 traversed += x->level[i].span;
217 x = x->level[i].forward;
218 }
219 update[i] = x;
220 }
221
222 traversed++;
223 x = x->level[0].forward;
224 while (x && traversed <= end) {
225 zskiplistNode *next = x->level[0].forward;
226 zslDeleteNode(zsl,x,update);
227 dictDelete(dict,x->obj);
228 zslFreeNode(x);
229 removed++;
230 traversed++;
231 x = next;
232 }
233 return removed;
234 }
235
236 /* Find the first node having a score equal or greater than the specified one.
237 * Returns NULL if there is no match. */
238 zskiplistNode *zslFirstWithScore(zskiplist *zsl, double score) {
239 zskiplistNode *x;
240 int i;
241
242 x = zsl->header;
243 for (i = zsl->level-1; i >= 0; i--) {
244 while (x->level[i].forward && x->level[i].forward->score < score)
245 x = x->level[i].forward;
246 }
247 /* We may have multiple elements with the same score, what we need
248 * is to find the element with both the right score and object. */
249 return x->level[0].forward;
250 }
251
252 /* Find the rank for an element by both score and key.
253 * Returns 0 when the element cannot be found, rank otherwise.
254 * Note that the rank is 1-based due to the span of zsl->header to the
255 * first element. */
256 unsigned long zslistTypeGetRank(zskiplist *zsl, double score, robj *o) {
257 zskiplistNode *x;
258 unsigned long rank = 0;
259 int i;
260
261 x = zsl->header;
262 for (i = zsl->level-1; i >= 0; i--) {
263 while (x->level[i].forward &&
264 (x->level[i].forward->score < score ||
265 (x->level[i].forward->score == score &&
266 compareStringObjects(x->level[i].forward->obj,o) <= 0))) {
267 rank += x->level[i].span;
268 x = x->level[i].forward;
269 }
270
271 /* x might be equal to zsl->header, so test if obj is non-NULL */
272 if (x->obj && equalStringObjects(x->obj,o)) {
273 return rank;
274 }
275 }
276 return 0;
277 }
278
279 /* Finds an element by its rank. The rank argument needs to be 1-based. */
280 zskiplistNode* zslistTypeGetElementByRank(zskiplist *zsl, unsigned long rank) {
281 zskiplistNode *x;
282 unsigned long traversed = 0;
283 int i;
284
285 x = zsl->header;
286 for (i = zsl->level-1; i >= 0; i--) {
287 while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
288 {
289 traversed += x->level[i].span;
290 x = x->level[i].forward;
291 }
292 if (traversed == rank) {
293 return x;
294 }
295 }
296 return NULL;
297 }
298
299 typedef struct {
300 double min, max;
301 int minex, maxex; /* are min or max exclusive? */
302 } zrangespec;
303
304 /* Populate the rangespec according to the objects min and max. */
305 int zslParseRange(robj *min, robj *max, zrangespec *spec) {
306 spec->minex = spec->maxex = 0;
307
308 /* Parse the min-max interval. If one of the values is prefixed
309 * by the "(" character, it's considered "open". For instance
310 * ZRANGEBYSCORE zset (1.5 (2.5 will match min < x < max
311 * ZRANGEBYSCORE zset 1.5 2.5 will instead match min <= x <= max */
312 if (min->encoding == REDIS_ENCODING_INT) {
313 spec->min = (long)min->ptr;
314 } else {
315 if (((char*)min->ptr)[0] == '(') {
316 spec->min = strtod((char*)min->ptr+1,NULL);
317 spec->minex = 1;
318 } else {
319 spec->min = strtod((char*)min->ptr,NULL);
320 }
321 }
322 if (max->encoding == REDIS_ENCODING_INT) {
323 spec->max = (long)max->ptr;
324 } else {
325 if (((char*)max->ptr)[0] == '(') {
326 spec->max = strtod((char*)max->ptr+1,NULL);
327 spec->maxex = 1;
328 } else {
329 spec->max = strtod((char*)max->ptr,NULL);
330 }
331 }
332
333 return REDIS_OK;
334 }
335
336
337 /*-----------------------------------------------------------------------------
338 * Sorted set commands
339 *----------------------------------------------------------------------------*/
340
341 /* This generic command implements both ZADD and ZINCRBY. */
342 void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double score, int incr) {
343 robj *zsetobj;
344 zset *zs;
345 zskiplistNode *znode;
346
347 zsetobj = lookupKeyWrite(c->db,key);
348 if (zsetobj == NULL) {
349 zsetobj = createZsetObject();
350 dbAdd(c->db,key,zsetobj);
351 } else {
352 if (zsetobj->type != REDIS_ZSET) {
353 addReply(c,shared.wrongtypeerr);
354 return;
355 }
356 }
357 zs = zsetobj->ptr;
358
359 /* Since both ZADD and ZINCRBY are implemented here, we need to increment
360 * the score first by the current score if ZINCRBY is called. */
361 if (incr) {
362 /* Read the old score. If the element was not present starts from 0 */
363 dictEntry *de = dictFind(zs->dict,ele);
364 if (de != NULL)
365 score += *(double*)dictGetEntryVal(de);
366
367 if (isnan(score)) {
368 addReplyError(c,"resulting score is not a number (NaN)");
369 /* Note that we don't need to check if the zset may be empty and
370 * should be removed here, as we can only obtain Nan as score if
371 * there was already an element in the sorted set. */
372 return;
373 }
374 }
375
376 /* We need to remove and re-insert the element when it was already present
377 * in the dictionary, to update the skiplist. Note that we delay adding a
378 * pointer to the score because we want to reference the score in the
379 * skiplist node. */
380 if (dictAdd(zs->dict,ele,NULL) == DICT_OK) {
381 dictEntry *de;
382
383 /* New element */
384 incrRefCount(ele); /* added to hash */
385 znode = zslInsert(zs->zsl,score,ele);
386 incrRefCount(ele); /* added to skiplist */
387
388 /* Update the score in the dict entry */
389 de = dictFind(zs->dict,ele);
390 redisAssert(de != NULL);
391 dictGetEntryVal(de) = &znode->score;
392 touchWatchedKey(c->db,c->argv[1]);
393 server.dirty++;
394 if (incr)
395 addReplyDouble(c,score);
396 else
397 addReply(c,shared.cone);
398 } else {
399 dictEntry *de;
400 robj *curobj;
401 double *curscore;
402 int deleted;
403
404 /* Update score */
405 de = dictFind(zs->dict,ele);
406 redisAssert(de != NULL);
407 curobj = dictGetEntryKey(de);
408 curscore = dictGetEntryVal(de);
409
410 /* When the score is updated, reuse the existing string object to
411 * prevent extra alloc/dealloc of strings on ZINCRBY. */
412 if (score != *curscore) {
413 deleted = zslDelete(zs->zsl,*curscore,curobj);
414 redisAssert(deleted != 0);
415 znode = zslInsert(zs->zsl,score,curobj);
416 incrRefCount(curobj);
417
418 /* Update the score in the current dict entry */
419 dictGetEntryVal(de) = &znode->score;
420 touchWatchedKey(c->db,c->argv[1]);
421 server.dirty++;
422 }
423 if (incr)
424 addReplyDouble(c,score);
425 else
426 addReply(c,shared.czero);
427 }
428 }
429
430 void zaddCommand(redisClient *c) {
431 double scoreval;
432 if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
433 zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
434 }
435
436 void zincrbyCommand(redisClient *c) {
437 double scoreval;
438 if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
439 zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
440 }
441
442 void zremCommand(redisClient *c) {
443 robj *zsetobj;
444 zset *zs;
445 dictEntry *de;
446 double curscore;
447 int deleted;
448
449 if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
450 checkType(c,zsetobj,REDIS_ZSET)) return;
451
452 zs = zsetobj->ptr;
453 de = dictFind(zs->dict,c->argv[2]);
454 if (de == NULL) {
455 addReply(c,shared.czero);
456 return;
457 }
458 /* Delete from the skiplist */
459 curscore = *(double*)dictGetEntryVal(de);
460 deleted = zslDelete(zs->zsl,curscore,c->argv[2]);
461 redisAssert(deleted != 0);
462
463 /* Delete from the hash table */
464 dictDelete(zs->dict,c->argv[2]);
465 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
466 if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
467 touchWatchedKey(c->db,c->argv[1]);
468 server.dirty++;
469 addReply(c,shared.cone);
470 }
471
472 void zremrangebyscoreCommand(redisClient *c) {
473 double min;
474 double max;
475 long deleted;
476 robj *zsetobj;
477 zset *zs;
478
479 if ((getDoubleFromObjectOrReply(c, c->argv[2], &min, NULL) != REDIS_OK) ||
480 (getDoubleFromObjectOrReply(c, c->argv[3], &max, NULL) != REDIS_OK)) return;
481
482 if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
483 checkType(c,zsetobj,REDIS_ZSET)) return;
484
485 zs = zsetobj->ptr;
486 deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict);
487 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
488 if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
489 if (deleted) touchWatchedKey(c->db,c->argv[1]);
490 server.dirty += deleted;
491 addReplyLongLong(c,deleted);
492 }
493
494 void zremrangebyrankCommand(redisClient *c) {
495 long start;
496 long end;
497 int llen;
498 long deleted;
499 robj *zsetobj;
500 zset *zs;
501
502 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
503 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
504
505 if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
506 checkType(c,zsetobj,REDIS_ZSET)) return;
507 zs = zsetobj->ptr;
508 llen = zs->zsl->length;
509
510 /* convert negative indexes */
511 if (start < 0) start = llen+start;
512 if (end < 0) end = llen+end;
513 if (start < 0) start = 0;
514
515 /* Invariant: start >= 0, so this test will be true when end < 0.
516 * The range is empty when start > end or start >= length. */
517 if (start > end || start >= llen) {
518 addReply(c,shared.czero);
519 return;
520 }
521 if (end >= llen) end = llen-1;
522
523 /* increment start and end because zsl*Rank functions
524 * use 1-based rank */
525 deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
526 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
527 if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
528 if (deleted) touchWatchedKey(c->db,c->argv[1]);
529 server.dirty += deleted;
530 addReplyLongLong(c, deleted);
531 }
532
533 typedef struct {
534 dict *dict;
535 double weight;
536 } zsetopsrc;
537
538 int qsortCompareZsetopsrcByCardinality(const void *s1, const void *s2) {
539 zsetopsrc *d1 = (void*) s1, *d2 = (void*) s2;
540 unsigned long size1, size2;
541 size1 = d1->dict ? dictSize(d1->dict) : 0;
542 size2 = d2->dict ? dictSize(d2->dict) : 0;
543 return size1 - size2;
544 }
545
546 #define REDIS_AGGR_SUM 1
547 #define REDIS_AGGR_MIN 2
548 #define REDIS_AGGR_MAX 3
549 #define zunionInterDictValue(_e) (dictGetEntryVal(_e) == NULL ? 1.0 : *(double*)dictGetEntryVal(_e))
550
551 inline static void zunionInterAggregate(double *target, double val, int aggregate) {
552 if (aggregate == REDIS_AGGR_SUM) {
553 *target = *target + val;
554 /* The result of adding two doubles is NaN when one variable
555 * is +inf and the other is -inf. When these numbers are added,
556 * we maintain the convention of the result being 0.0. */
557 if (isnan(*target)) *target = 0.0;
558 } else if (aggregate == REDIS_AGGR_MIN) {
559 *target = val < *target ? val : *target;
560 } else if (aggregate == REDIS_AGGR_MAX) {
561 *target = val > *target ? val : *target;
562 } else {
563 /* safety net */
564 redisPanic("Unknown ZUNION/INTER aggregate type");
565 }
566 }
567
568 void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
569 int i, j, setnum;
570 int aggregate = REDIS_AGGR_SUM;
571 zsetopsrc *src;
572 robj *dstobj;
573 zset *dstzset;
574 zskiplistNode *znode;
575 dictIterator *di;
576 dictEntry *de;
577 int touched = 0;
578
579 /* expect setnum input keys to be given */
580 setnum = atoi(c->argv[2]->ptr);
581 if (setnum < 1) {
582 addReplyError(c,
583 "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE");
584 return;
585 }
586
587 /* test if the expected number of keys would overflow */
588 if (3+setnum > c->argc) {
589 addReply(c,shared.syntaxerr);
590 return;
591 }
592
593 /* read keys to be used for input */
594 src = zmalloc(sizeof(zsetopsrc) * setnum);
595 for (i = 0, j = 3; i < setnum; i++, j++) {
596 robj *obj = lookupKeyWrite(c->db,c->argv[j]);
597 if (!obj) {
598 src[i].dict = NULL;
599 } else {
600 if (obj->type == REDIS_ZSET) {
601 src[i].dict = ((zset*)obj->ptr)->dict;
602 } else if (obj->type == REDIS_SET) {
603 src[i].dict = (obj->ptr);
604 } else {
605 zfree(src);
606 addReply(c,shared.wrongtypeerr);
607 return;
608 }
609 }
610
611 /* default all weights to 1 */
612 src[i].weight = 1.0;
613 }
614
615 /* parse optional extra arguments */
616 if (j < c->argc) {
617 int remaining = c->argc - j;
618
619 while (remaining) {
620 if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
621 j++; remaining--;
622 for (i = 0; i < setnum; i++, j++, remaining--) {
623 if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
624 "weight value is not a double") != REDIS_OK)
625 {
626 zfree(src);
627 return;
628 }
629 }
630 } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
631 j++; remaining--;
632 if (!strcasecmp(c->argv[j]->ptr,"sum")) {
633 aggregate = REDIS_AGGR_SUM;
634 } else if (!strcasecmp(c->argv[j]->ptr,"min")) {
635 aggregate = REDIS_AGGR_MIN;
636 } else if (!strcasecmp(c->argv[j]->ptr,"max")) {
637 aggregate = REDIS_AGGR_MAX;
638 } else {
639 zfree(src);
640 addReply(c,shared.syntaxerr);
641 return;
642 }
643 j++; remaining--;
644 } else {
645 zfree(src);
646 addReply(c,shared.syntaxerr);
647 return;
648 }
649 }
650 }
651
652 /* sort sets from the smallest to largest, this will improve our
653 * algorithm's performance */
654 qsort(src,setnum,sizeof(zsetopsrc),qsortCompareZsetopsrcByCardinality);
655
656 dstobj = createZsetObject();
657 dstzset = dstobj->ptr;
658
659 if (op == REDIS_OP_INTER) {
660 /* skip going over all entries if the smallest zset is NULL or empty */
661 if (src[0].dict && dictSize(src[0].dict) > 0) {
662 /* precondition: as src[0].dict is non-empty and the zsets are ordered
663 * from small to large, all src[i > 0].dict are non-empty too */
664 di = dictGetIterator(src[0].dict);
665 while((de = dictNext(di)) != NULL) {
666 double *score = zmalloc(sizeof(double)), value;
667 *score = src[0].weight * zunionInterDictValue(de);
668
669 for (j = 1; j < setnum; j++) {
670 dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
671 if (other) {
672 value = src[j].weight * zunionInterDictValue(other);
673 zunionInterAggregate(score, value, aggregate);
674 } else {
675 break;
676 }
677 }
678
679 /* skip entry when not present in every source dict */
680 if (j != setnum) {
681 zfree(score);
682 } else {
683 robj *o = dictGetEntryKey(de);
684 znode = zslInsert(dstzset->zsl,*score,o);
685 incrRefCount(o); /* added to skiplist */
686 dictAdd(dstzset->dict,o,&znode->score);
687 incrRefCount(o); /* added to dictionary */
688 }
689 }
690 dictReleaseIterator(di);
691 }
692 } else if (op == REDIS_OP_UNION) {
693 for (i = 0; i < setnum; i++) {
694 if (!src[i].dict) continue;
695
696 di = dictGetIterator(src[i].dict);
697 while((de = dictNext(di)) != NULL) {
698 /* skip key when already processed */
699 if (dictFind(dstzset->dict,dictGetEntryKey(de)) != NULL) continue;
700
701 double *score = zmalloc(sizeof(double)), value;
702 *score = src[i].weight * zunionInterDictValue(de);
703
704 /* because the zsets are sorted by size, its only possible
705 * for sets at larger indices to hold this entry */
706 for (j = (i+1); j < setnum; j++) {
707 dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
708 if (other) {
709 value = src[j].weight * zunionInterDictValue(other);
710 zunionInterAggregate(score, value, aggregate);
711 }
712 }
713
714 robj *o = dictGetEntryKey(de);
715 znode = zslInsert(dstzset->zsl,*score,o);
716 incrRefCount(o); /* added to skiplist */
717 dictAdd(dstzset->dict,o,&znode->score);
718 incrRefCount(o); /* added to dictionary */
719 }
720 dictReleaseIterator(di);
721 }
722 } else {
723 /* unknown operator */
724 redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
725 }
726
727 if (dbDelete(c->db,dstkey)) {
728 touchWatchedKey(c->db,dstkey);
729 touched = 1;
730 server.dirty++;
731 }
732 if (dstzset->zsl->length) {
733 dbAdd(c->db,dstkey,dstobj);
734 addReplyLongLong(c, dstzset->zsl->length);
735 if (!touched) touchWatchedKey(c->db,dstkey);
736 server.dirty++;
737 } else {
738 decrRefCount(dstobj);
739 addReply(c, shared.czero);
740 }
741 zfree(src);
742 }
743
744 void zunionstoreCommand(redisClient *c) {
745 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_UNION);
746 }
747
748 void zinterstoreCommand(redisClient *c) {
749 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_INTER);
750 }
751
752 void zrangeGenericCommand(redisClient *c, int reverse) {
753 robj *o;
754 long start;
755 long end;
756 int withscores = 0;
757 int llen;
758 int rangelen, j;
759 zset *zsetobj;
760 zskiplist *zsl;
761 zskiplistNode *ln;
762 robj *ele;
763
764 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
765 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
766
767 if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
768 withscores = 1;
769 } else if (c->argc >= 5) {
770 addReply(c,shared.syntaxerr);
771 return;
772 }
773
774 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
775 || checkType(c,o,REDIS_ZSET)) return;
776 zsetobj = o->ptr;
777 zsl = zsetobj->zsl;
778 llen = zsl->length;
779
780 /* convert negative indexes */
781 if (start < 0) start = llen+start;
782 if (end < 0) end = llen+end;
783 if (start < 0) start = 0;
784
785 /* Invariant: start >= 0, so this test will be true when end < 0.
786 * The range is empty when start > end or start >= length. */
787 if (start > end || start >= llen) {
788 addReply(c,shared.emptymultibulk);
789 return;
790 }
791 if (end >= llen) end = llen-1;
792 rangelen = (end-start)+1;
793
794 /* check if starting point is trivial, before searching
795 * the element in log(N) time */
796 if (reverse) {
797 ln = start == 0 ? zsl->tail : zslistTypeGetElementByRank(zsl, llen-start);
798 } else {
799 ln = start == 0 ?
800 zsl->header->level[0].forward : zslistTypeGetElementByRank(zsl, start+1);
801 }
802
803 /* Return the result in form of a multi-bulk reply */
804 addReplyMultiBulkLen(c,withscores ? (rangelen*2) : rangelen);
805 for (j = 0; j < rangelen; j++) {
806 ele = ln->obj;
807 addReplyBulk(c,ele);
808 if (withscores)
809 addReplyDouble(c,ln->score);
810 ln = reverse ? ln->backward : ln->level[0].forward;
811 }
812 }
813
814 void zrangeCommand(redisClient *c) {
815 zrangeGenericCommand(c,0);
816 }
817
818 void zrevrangeCommand(redisClient *c) {
819 zrangeGenericCommand(c,1);
820 }
821
822 /* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE and ZCOUNT.
823 * If "justcount", only the number of elements in the range is returned. */
824 void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) {
825 zrangespec range;
826 robj *o, *emptyreply;
827 zset *zsetobj;
828 zskiplist *zsl;
829 zskiplistNode *ln;
830 int offset = 0, limit = -1;
831 int withscores = 0;
832 unsigned long rangelen = 0;
833 void *replylen = NULL;
834
835 /* Parse the range arguments. */
836 zslParseRange(c->argv[2],c->argv[3],&range);
837
838 /* Parse optional extra arguments. Note that ZCOUNT will exactly have
839 * 4 arguments, so we'll never enter the following code path. */
840 if (c->argc > 4) {
841 int remaining = c->argc - 4;
842 int pos = 4;
843
844 while (remaining) {
845 if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) {
846 pos++; remaining--;
847 withscores = 1;
848 } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {
849 offset = atoi(c->argv[pos+1]->ptr);
850 limit = atoi(c->argv[pos+2]->ptr);
851 pos += 3; remaining -= 3;
852 } else {
853 addReply(c,shared.syntaxerr);
854 return;
855 }
856 }
857 }
858
859 /* Ok, lookup the key and get the range */
860 emptyreply = justcount ? shared.czero : shared.emptymultibulk;
861 if ((o = lookupKeyReadOrReply(c,c->argv[1],emptyreply)) == NULL ||
862 checkType(c,o,REDIS_ZSET)) return;
863 zsetobj = o->ptr;
864 zsl = zsetobj->zsl;
865
866 /* If reversed, assume the elements are sorted from high to low score. */
867 ln = zslFirstWithScore(zsl,range.min);
868 if (reverse) {
869 /* If range.min is out of range, ln will be NULL and we need to use
870 * the tail of the skiplist as first node of the range. */
871 if (ln == NULL) ln = zsl->tail;
872
873 /* zslFirstWithScore returns the first element with where with
874 * score >= range.min, so backtrack to make sure the element we use
875 * here has score <= range.min. */
876 while (ln && ln->score > range.min) ln = ln->backward;
877
878 /* Move to the right element according to the range spec. */
879 if (range.minex) {
880 /* Find last element with score < range.min */
881 while (ln && ln->score == range.min) ln = ln->backward;
882 } else {
883 /* Find last element with score <= range.min */
884 while (ln && ln->level[0].forward &&
885 ln->level[0].forward->score == range.min)
886 ln = ln->level[0].forward;
887 }
888 } else {
889 if (range.minex) {
890 /* Find first element with score > range.min */
891 while (ln && ln->score == range.min) ln = ln->level[0].forward;
892 }
893 }
894
895 /* No "first" element in the specified interval. */
896 if (ln == NULL) {
897 addReply(c,emptyreply);
898 return;
899 }
900
901 /* We don't know in advance how many matching elements there
902 * are in the list, so we push this object that will represent
903 * the multi-bulk length in the output buffer, and will "fix"
904 * it later */
905 if (!justcount)
906 replylen = addDeferredMultiBulkLength(c);
907
908 /* If there is an offset, just traverse the number of elements without
909 * checking the score because that is done in the next loop. */
910 while(ln && offset--) {
911 if (reverse)
912 ln = ln->backward;
913 else
914 ln = ln->level[0].forward;
915 }
916
917 while (ln && limit--) {
918 /* Check if this this element is in range. */
919 if (reverse) {
920 if (range.maxex) {
921 /* Element should have score > range.max */
922 if (ln->score <= range.max) break;
923 } else {
924 /* Element should have score >= range.max */
925 if (ln->score < range.max) break;
926 }
927 } else {
928 if (range.maxex) {
929 /* Element should have score < range.max */
930 if (ln->score >= range.max) break;
931 } else {
932 /* Element should have score <= range.max */
933 if (ln->score > range.max) break;
934 }
935 }
936
937 /* Do our magic */
938 rangelen++;
939 if (!justcount) {
940 addReplyBulk(c,ln->obj);
941 if (withscores)
942 addReplyDouble(c,ln->score);
943 }
944
945 if (reverse)
946 ln = ln->backward;
947 else
948 ln = ln->level[0].forward;
949 }
950
951 if (justcount) {
952 addReplyLongLong(c,(long)rangelen);
953 } else {
954 setDeferredMultiBulkLength(c,replylen,
955 withscores ? (rangelen*2) : rangelen);
956 }
957 }
958
959 void zrangebyscoreCommand(redisClient *c) {
960 genericZrangebyscoreCommand(c,0,0);
961 }
962
963 void zrevrangebyscoreCommand(redisClient *c) {
964 genericZrangebyscoreCommand(c,1,0);
965 }
966
967 void zcountCommand(redisClient *c) {
968 genericZrangebyscoreCommand(c,0,1);
969 }
970
971 void zcardCommand(redisClient *c) {
972 robj *o;
973 zset *zs;
974
975 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
976 checkType(c,o,REDIS_ZSET)) return;
977
978 zs = o->ptr;
979 addReplyLongLong(c,zs->zsl->length);
980 }
981
982 void zscoreCommand(redisClient *c) {
983 robj *o;
984 zset *zs;
985 dictEntry *de;
986
987 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
988 checkType(c,o,REDIS_ZSET)) return;
989
990 zs = o->ptr;
991 de = dictFind(zs->dict,c->argv[2]);
992 if (!de) {
993 addReply(c,shared.nullbulk);
994 } else {
995 double *score = dictGetEntryVal(de);
996
997 addReplyDouble(c,*score);
998 }
999 }
1000
1001 void zrankGenericCommand(redisClient *c, int reverse) {
1002 robj *o;
1003 zset *zs;
1004 zskiplist *zsl;
1005 dictEntry *de;
1006 unsigned long rank;
1007 double *score;
1008
1009 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
1010 checkType(c,o,REDIS_ZSET)) return;
1011
1012 zs = o->ptr;
1013 zsl = zs->zsl;
1014 de = dictFind(zs->dict,c->argv[2]);
1015 if (!de) {
1016 addReply(c,shared.nullbulk);
1017 return;
1018 }
1019
1020 score = dictGetEntryVal(de);
1021 rank = zslistTypeGetRank(zsl, *score, c->argv[2]);
1022 if (rank) {
1023 if (reverse) {
1024 addReplyLongLong(c, zsl->length - rank);
1025 } else {
1026 addReplyLongLong(c, rank-1);
1027 }
1028 } else {
1029 addReply(c,shared.nullbulk);
1030 }
1031 }
1032
1033 void zrankCommand(redisClient *c) {
1034 zrankGenericCommand(c, 0);
1035 }
1036
1037 void zrevrankCommand(redisClient *c) {
1038 zrankGenericCommand(c, 1);
1039 }