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