1/* SORT command and helper functions.
2 *
3 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
4 * All rights reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions are met:
8 *
9 * * Redistributions of source code must retain the above copyright notice,
10 * this list of conditions and the following disclaimer.
11 * * Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 * * Neither the name of Redis nor the names of its contributors may be used
15 * to endorse or promote products derived from this software without
16 * specific prior written permission.
17 *
18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
19 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
22 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
23 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
25 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
26 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
28 * POSSIBILITY OF SUCH DAMAGE.
29 */
30
31
32#include "server.h"
33#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
34#include <math.h> /* isnan() */
35
36zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank);
37
38redisSortOperation *createSortOperation(int type, robj *pattern) {
39 redisSortOperation *so = zmalloc(sizeof(*so));
40 so->type = type;
41 so->pattern = pattern;
42 return so;
43}
44
45/* Return the value associated to the key with a name obtained using
46 * the following rules:
47 *
48 * 1) The first occurrence of '*' in 'pattern' is substituted with 'subst'.
49 *
50 * 2) If 'pattern' matches the "->" string, everything on the left of
51 * the arrow is treated as the name of a hash field, and the part on the
52 * left as the key name containing a hash. The value of the specified
53 * field is returned.
54 *
55 * 3) If 'pattern' equals "#", the function simply returns 'subst' itself so
56 * that the SORT command can be used like: SORT key GET # to retrieve
57 * the Set/List elements directly.
58 *
59 * The returned object will always have its refcount increased by 1
60 * when it is non-NULL. */
61robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
62 char *p, *f, *k;
63 sds spat, ssub;
64 robj *keyobj, *fieldobj = NULL, *o;
65 int prefixlen, sublen, postfixlen, fieldlen;
66
67 /* If the pattern is "#" return the substitution object itself in order
68 * to implement the "SORT ... GET #" feature. */
69 spat = pattern->ptr;
70 if (spat[0] == '#' && spat[1] == '\0') {
71 incrRefCount(subst);
72 return subst;
73 }
74
75 /* The substitution object may be specially encoded. If so we create
76 * a decoded object on the fly. Otherwise getDecodedObject will just
77 * increment the ref count, that we'll decrement later. */
78 subst = getDecodedObject(subst);
79 ssub = subst->ptr;
80
81 /* If we can't find '*' in the pattern we return NULL as to GET a
82 * fixed key does not make sense. */
83 p = strchr(spat,'*');
84 if (!p) {
85 decrRefCount(subst);
86 return NULL;
87 }
88
89 /* Find out if we're dealing with a hash dereference. */
90 if ((f = strstr(p+1, "->")) != NULL && *(f+2) != '\0') {
91 fieldlen = sdslen(spat)-(f-spat)-2;
92 fieldobj = createStringObject(f+2,fieldlen);
93 } else {
94 fieldlen = 0;
95 }
96
97 /* Perform the '*' substitution. */
98 prefixlen = p-spat;
99 sublen = sdslen(ssub);
100 postfixlen = sdslen(spat)-(prefixlen+1)-(fieldlen ? fieldlen+2 : 0);
101 keyobj = createStringObject(NULL,prefixlen+sublen+postfixlen);
102 k = keyobj->ptr;
103 memcpy(k,spat,prefixlen);
104 memcpy(k+prefixlen,ssub,sublen);
105 memcpy(k+prefixlen+sublen,p+1,postfixlen);
106 decrRefCount(subst); /* Incremented by decodeObject() */
107
108 /* Lookup substituted key */
109 o = lookupKeyRead(db, keyobj);
110 if (o == NULL) goto noobj;
111
112 if (fieldobj) {
113 if (o->type != OBJ_HASH) goto noobj;
114
115 /* Retrieve value from hash by the field name. The returned object
116 * is a new object with refcount already incremented. */
117 o = hashTypeGetValueObject(o, fieldobj->ptr);
118 } else {
119 if (o->type != OBJ_STRING) goto noobj;
120
121 /* Every object that this function returns needs to have its refcount
122 * increased. sortCommand decreases it again. */
123 incrRefCount(o);
124 }
125 decrRefCount(keyobj);
126 if (fieldobj) decrRefCount(fieldobj);
127 return o;
128
129noobj:
130 decrRefCount(keyobj);
131 if (fieldlen) decrRefCount(fieldobj);
132 return NULL;
133}
134
135/* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
136 * the additional parameter is not standard but a BSD-specific we have to
137 * pass sorting parameters via the global 'server' structure */
138int sortCompare(const void *s1, const void *s2) {
139 const redisSortObject *so1 = s1, *so2 = s2;
140 int cmp;
141
142 if (!server.sort_alpha) {
143 /* Numeric sorting. Here it's trivial as we precomputed scores */
144 if (so1->u.score > so2->u.score) {
145 cmp = 1;
146 } else if (so1->u.score < so2->u.score) {
147 cmp = -1;
148 } else {
149 /* Objects have the same score, but we don't want the comparison
150 * to be undefined, so we compare objects lexicographically.
151 * This way the result of SORT is deterministic. */
152 cmp = compareStringObjects(so1->obj,so2->obj);
153 }
154 } else {
155 /* Alphanumeric sorting */
156 if (server.sort_bypattern) {
157 if (!so1->u.cmpobj || !so2->u.cmpobj) {
158 /* At least one compare object is NULL */
159 if (so1->u.cmpobj == so2->u.cmpobj)
160 cmp = 0;
161 else if (so1->u.cmpobj == NULL)
162 cmp = -1;
163 else
164 cmp = 1;
165 } else {
166 /* We have both the objects, compare them. */
167 if (server.sort_store) {
168 cmp = compareStringObjects(so1->u.cmpobj,so2->u.cmpobj);
169 } else {
170 /* Here we can use strcoll() directly as we are sure that
171 * the objects are decoded string objects. */
172 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
173 }
174 }
175 } else {
176 /* Compare elements directly. */
177 if (server.sort_store) {
178 cmp = compareStringObjects(so1->obj,so2->obj);
179 } else {
180 cmp = collateStringObjects(so1->obj,so2->obj);
181 }
182 }
183 }
184 return server.sort_desc ? -cmp : cmp;
185}
186
187/* The SORT command is the most complex command in Redis. Warning: this code
188 * is optimized for speed and a bit less for readability */
189void sortCommandGeneric(client *c, int readonly) {
190 list *operations;
191 unsigned int outputlen = 0;
192 int desc = 0, alpha = 0;
193 long limit_start = 0, limit_count = -1, start, end;
194 int j, dontsort = 0, vectorlen;
195 int getop = 0; /* GET operation counter */
196 int int_conversion_error = 0;
197 int syntax_error = 0;
198 robj *sortval, *sortby = NULL, *storekey = NULL;
199 redisSortObject *vector; /* Resulting vector to sort */
200 int user_has_full_key_access = 0; /* ACL - used in order to verify 'get' and 'by' options can be used */
201 /* Create a list of operations to perform for every sorted element.
202 * Operations can be GET */
203 operations = listCreate();
204 listSetFreeMethod(operations,zfree);
205 j = 2; /* options start at argv[2] */
206
207 user_has_full_key_access = ACLUserCheckCmdWithUnrestrictedKeyAccess(c->user, c->cmd, c->argv, c->argc, CMD_KEY_ACCESS);
208
209 /* The SORT command has an SQL-alike syntax, parse it */
210 while(j < c->argc) {
211 int leftargs = c->argc-j-1;
212 if (!strcasecmp(c->argv[j]->ptr,"asc")) {
213 desc = 0;
214 } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
215 desc = 1;
216 } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
217 alpha = 1;
218 } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
219 if ((getLongFromObjectOrReply(c, c->argv[j+1], &limit_start, NULL)
220 != C_OK) ||
221 (getLongFromObjectOrReply(c, c->argv[j+2], &limit_count, NULL)
222 != C_OK))
223 {
224 syntax_error++;
225 break;
226 }
227 j+=2;
228 } else if (readonly == 0 && !strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
229 storekey = c->argv[j+1];
230 j++;
231 } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
232 sortby = c->argv[j+1];
233 /* If the BY pattern does not contain '*', i.e. it is constant,
234 * we don't need to sort nor to lookup the weight keys. */
235 if (strchr(c->argv[j+1]->ptr,'*') == NULL) {
236 dontsort = 1;
237 } else {
238 /* If BY is specified with a real pattern, we can't accept
239 * it in cluster mode. */
240 if (server.cluster_enabled) {
241 addReplyError(c,"BY option of SORT denied in Cluster mode.");
242 syntax_error++;
243 break;
244 }
245 /* If BY is specified with a real pattern, we can't accept
246 * it if no full ACL key access is applied for this command. */
247 if (!user_has_full_key_access) {
248 addReplyError(c,"BY option of SORT denied due to insufficient ACL permissions.");
249 syntax_error++;
250 break;
251 }
252 }
253 j++;
254 } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
255 if (server.cluster_enabled) {
256 addReplyError(c,"GET option of SORT denied in Cluster mode.");
257 syntax_error++;
258 break;
259 }
260 if (!user_has_full_key_access) {
261 addReplyError(c,"GET option of SORT denied due to insufficient ACL permissions.");
262 syntax_error++;
263 break;
264 }
265 listAddNodeTail(operations,createSortOperation(
266 SORT_OP_GET,c->argv[j+1]));
267 getop++;
268 j++;
269 } else {
270 addReplyErrorObject(c,shared.syntaxerr);
271 syntax_error++;
272 break;
273 }
274 j++;
275 }
276
277 /* Handle syntax errors set during options parsing. */
278 if (syntax_error) {
279 listRelease(operations);
280 return;
281 }
282
283 /* Lookup the key to sort. It must be of the right types */
284 sortval = lookupKeyRead(c->db, c->argv[1]);
285 if (sortval && sortval->type != OBJ_SET &&
286 sortval->type != OBJ_LIST &&
287 sortval->type != OBJ_ZSET)
288 {
289 listRelease(operations);
290 addReplyErrorObject(c,shared.wrongtypeerr);
291 return;
292 }
293
294 /* Now we need to protect sortval incrementing its count, in the future
295 * SORT may have options able to overwrite/delete keys during the sorting
296 * and the sorted key itself may get destroyed */
297 if (sortval)
298 incrRefCount(sortval);
299 else
300 sortval = createQuicklistObject();
301
302
303 /* When sorting a set with no sort specified, we must sort the output
304 * so the result is consistent across scripting and replication.
305 *
306 * The other types (list, sorted set) will retain their native order
307 * even if no sort order is requested, so they remain stable across
308 * scripting and replication. */
309 if (dontsort &&
310 sortval->type == OBJ_SET &&
311 (storekey || c->flags & CLIENT_SCRIPT))
312 {
313 /* Force ALPHA sorting */
314 dontsort = 0;
315 alpha = 1;
316 sortby = NULL;
317 }
318
319 /* Destructively convert encoded sorted sets for SORT. */
320 if (sortval->type == OBJ_ZSET)
321 zsetConvert(sortval, OBJ_ENCODING_SKIPLIST);
322
323 /* Obtain the length of the object to sort. */
324 switch(sortval->type) {
325 case OBJ_LIST: vectorlen = listTypeLength(sortval); break;
326 case OBJ_SET: vectorlen = setTypeSize(sortval); break;
327 case OBJ_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
328 default: vectorlen = 0; serverPanic("Bad SORT type"); /* Avoid GCC warning */
329 }
330
331 /* Perform LIMIT start,count sanity checking. */
332 start = (limit_start < 0) ? 0 : limit_start;
333 end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
334 if (start >= vectorlen) {
335 start = vectorlen-1;
336 end = vectorlen-2;
337 }
338 if (end >= vectorlen) end = vectorlen-1;
339
340 /* Whenever possible, we load elements into the output array in a more
341 * direct way. This is possible if:
342 *
343 * 1) The object to sort is a sorted set or a list (internally sorted).
344 * 2) There is nothing to sort as dontsort is true (BY <constant string>).
345 *
346 * In this special case, if we have a LIMIT option that actually reduces
347 * the number of elements to fetch, we also optimize to just load the
348 * range we are interested in and allocating a vector that is big enough
349 * for the selected range length. */
350 if ((sortval->type == OBJ_ZSET || sortval->type == OBJ_LIST) &&
351 dontsort &&
352 (start != 0 || end != vectorlen-1))
353 {
354 vectorlen = end-start+1;
355 }
356
357 /* Load the sorting vector with all the objects to sort */
358 vector = zmalloc(sizeof(redisSortObject)*vectorlen);
359 j = 0;
360
361 if (sortval->type == OBJ_LIST && dontsort) {
362 /* Special handling for a list, if 'dontsort' is true.
363 * This makes sure we return elements in the list original
364 * ordering, accordingly to DESC / ASC options.
365 *
366 * Note that in this case we also handle LIMIT here in a direct
367 * way, just getting the required range, as an optimization. */
368 if (end >= start) {
369 listTypeIterator *li;
370 listTypeEntry entry;
371 li = listTypeInitIterator(sortval,
372 desc ? (long)(listTypeLength(sortval) - start - 1) : start,
373 desc ? LIST_HEAD : LIST_TAIL);
374
375 while(j < vectorlen && listTypeNext(li,&entry)) {
376 vector[j].obj = listTypeGet(&entry);
377 vector[j].u.score = 0;
378 vector[j].u.cmpobj = NULL;
379 j++;
380 }
381 listTypeReleaseIterator(li);
382 /* Fix start/end: output code is not aware of this optimization. */
383 end -= start;
384 start = 0;
385 }
386 } else if (sortval->type == OBJ_LIST) {
387 listTypeIterator *li = listTypeInitIterator(sortval,0,LIST_TAIL);
388 listTypeEntry entry;
389 while(listTypeNext(li,&entry)) {
390 vector[j].obj = listTypeGet(&entry);
391 vector[j].u.score = 0;
392 vector[j].u.cmpobj = NULL;
393 j++;
394 }
395 listTypeReleaseIterator(li);
396 } else if (sortval->type == OBJ_SET) {
397 setTypeIterator *si = setTypeInitIterator(sortval);
398 sds sdsele;
399 while((sdsele = setTypeNextObject(si)) != NULL) {
400 vector[j].obj = createObject(OBJ_STRING,sdsele);
401 vector[j].u.score = 0;
402 vector[j].u.cmpobj = NULL;
403 j++;
404 }
405 setTypeReleaseIterator(si);
406 } else if (sortval->type == OBJ_ZSET && dontsort) {
407 /* Special handling for a sorted set, if 'dontsort' is true.
408 * This makes sure we return elements in the sorted set original
409 * ordering, accordingly to DESC / ASC options.
410 *
411 * Note that in this case we also handle LIMIT here in a direct
412 * way, just getting the required range, as an optimization. */
413
414 zset *zs = sortval->ptr;
415 zskiplist *zsl = zs->zsl;
416 zskiplistNode *ln;
417 sds sdsele;
418 int rangelen = vectorlen;
419
420 /* Check if starting point is trivial, before doing log(N) lookup. */
421 if (desc) {
422 long zsetlen = dictSize(((zset*)sortval->ptr)->dict);
423
424 ln = zsl->tail;
425 if (start > 0)
426 ln = zslGetElementByRank(zsl,zsetlen-start);
427 } else {
428 ln = zsl->header->level[0].forward;
429 if (start > 0)
430 ln = zslGetElementByRank(zsl,start+1);
431 }
432
433 while(rangelen--) {
434 serverAssertWithInfo(c,sortval,ln != NULL);
435 sdsele = ln->ele;
436 vector[j].obj = createStringObject(sdsele,sdslen(sdsele));
437 vector[j].u.score = 0;
438 vector[j].u.cmpobj = NULL;
439 j++;
440 ln = desc ? ln->backward : ln->level[0].forward;
441 }
442 /* Fix start/end: output code is not aware of this optimization. */
443 end -= start;
444 start = 0;
445 } else if (sortval->type == OBJ_ZSET) {
446 dict *set = ((zset*)sortval->ptr)->dict;
447 dictIterator *di;
448 dictEntry *setele;
449 sds sdsele;
450 di = dictGetIterator(set);
451 while((setele = dictNext(di)) != NULL) {
452 sdsele = dictGetKey(setele);
453 vector[j].obj = createStringObject(sdsele,sdslen(sdsele));
454 vector[j].u.score = 0;
455 vector[j].u.cmpobj = NULL;
456 j++;
457 }
458 dictReleaseIterator(di);
459 } else {
460 serverPanic("Unknown type");
461 }
462 serverAssertWithInfo(c,sortval,j == vectorlen);
463
464 /* Now it's time to load the right scores in the sorting vector */
465 if (!dontsort) {
466 for (j = 0; j < vectorlen; j++) {
467 robj *byval;
468 if (sortby) {
469 /* lookup value to sort by */
470 byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
471 if (!byval) continue;
472 } else {
473 /* use object itself to sort by */
474 byval = vector[j].obj;
475 }
476
477 if (alpha) {
478 if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
479 } else {
480 if (sdsEncodedObject(byval)) {
481 char *eptr;
482
483 vector[j].u.score = strtod(byval->ptr,&eptr);
484 if (eptr[0] != '\0' || errno == ERANGE ||
485 isnan(vector[j].u.score))
486 {
487 int_conversion_error = 1;
488 }
489 } else if (byval->encoding == OBJ_ENCODING_INT) {
490 /* Don't need to decode the object if it's
491 * integer-encoded (the only encoding supported) so
492 * far. We can just cast it */
493 vector[j].u.score = (long)byval->ptr;
494 } else {
495 serverAssertWithInfo(c,sortval,1 != 1);
496 }
497 }
498
499 /* when the object was retrieved using lookupKeyByPattern,
500 * its refcount needs to be decreased. */
501 if (sortby) {
502 decrRefCount(byval);
503 }
504 }
505
506 server.sort_desc = desc;
507 server.sort_alpha = alpha;
508 server.sort_bypattern = sortby ? 1 : 0;
509 server.sort_store = storekey ? 1 : 0;
510 if (sortby && (start != 0 || end != vectorlen-1))
511 pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
512 else
513 qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
514 }
515
516 /* Send command output to the output buffer, performing the specified
517 * GET/DEL/INCR/DECR operations if any. */
518 outputlen = getop ? getop*(end-start+1) : end-start+1;
519 if (int_conversion_error) {
520 addReplyError(c,"One or more scores can't be converted into double");
521 } else if (storekey == NULL) {
522 /* STORE option not specified, sent the sorting result to client */
523 addReplyArrayLen(c,outputlen);
524 for (j = start; j <= end; j++) {
525 listNode *ln;
526 listIter li;
527
528 if (!getop) addReplyBulk(c,vector[j].obj);
529 listRewind(operations,&li);
530 while((ln = listNext(&li))) {
531 redisSortOperation *sop = ln->value;
532 robj *val = lookupKeyByPattern(c->db,sop->pattern,
533 vector[j].obj);
534
535 if (sop->type == SORT_OP_GET) {
536 if (!val) {
537 addReplyNull(c);
538 } else {
539 addReplyBulk(c,val);
540 decrRefCount(val);
541 }
542 } else {
543 /* Always fails */
544 serverAssertWithInfo(c,sortval,sop->type == SORT_OP_GET);
545 }
546 }
547 }
548 } else {
549 robj *sobj = createQuicklistObject();
550
551 /* STORE option specified, set the sorting result as a List object */
552 for (j = start; j <= end; j++) {
553 listNode *ln;
554 listIter li;
555
556 if (!getop) {
557 listTypePush(sobj,vector[j].obj,LIST_TAIL);
558 } else {
559 listRewind(operations,&li);
560 while((ln = listNext(&li))) {
561 redisSortOperation *sop = ln->value;
562 robj *val = lookupKeyByPattern(c->db,sop->pattern,
563 vector[j].obj);
564
565 if (sop->type == SORT_OP_GET) {
566 if (!val) val = createStringObject("",0);
567
568 /* listTypePush does an incrRefCount, so we should take care
569 * care of the incremented refcount caused by either
570 * lookupKeyByPattern or createStringObject("",0) */
571 listTypePush(sobj,val,LIST_TAIL);
572 decrRefCount(val);
573 } else {
574 /* Always fails */
575 serverAssertWithInfo(c,sortval,sop->type == SORT_OP_GET);
576 }
577 }
578 }
579 }
580 if (outputlen) {
581 setKey(c,c->db,storekey,sobj,0);
582 notifyKeyspaceEvent(NOTIFY_LIST,"sortstore",storekey,
583 c->db->id);
584 server.dirty += outputlen;
585 } else if (dbDelete(c->db,storekey)) {
586 signalModifiedKey(c,c->db,storekey);
587 notifyKeyspaceEvent(NOTIFY_GENERIC,"del",storekey,c->db->id);
588 server.dirty++;
589 }
590 decrRefCount(sobj);
591 addReplyLongLong(c,outputlen);
592 }
593
594 /* Cleanup */
595 for (j = 0; j < vectorlen; j++)
596 decrRefCount(vector[j].obj);
597
598 decrRefCount(sortval);
599 listRelease(operations);
600 for (j = 0; j < vectorlen; j++) {
601 if (alpha && vector[j].u.cmpobj)
602 decrRefCount(vector[j].u.cmpobj);
603 }
604 zfree(vector);
605}
606
607/* SORT wrapper function for read-only mode. */
608void sortroCommand(client *c) {
609 sortCommandGeneric(c, 1);
610}
611
612void sortCommand(client *c) {
613 sortCommandGeneric(c, 0);
614}
615