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 | |
36 | zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank); |
37 | |
38 | redisSortOperation *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. */ |
61 | robj *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 | |
129 | noobj: |
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 */ |
138 | int 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 */ |
189 | void 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. */ |
608 | void sortroCommand(client *c) { |
609 | sortCommandGeneric(c, 1); |
610 | } |
611 | |
612 | void sortCommand(client *c) { |
613 | sortCommandGeneric(c, 0); |
614 | } |
615 | |