]> git.saurik.com Git - redis.git/blob - src/sort.c
A reimplementation of blocking operation internals.
[redis.git] / src / sort.c
1 #include "redis.h"
2 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
3 #include <math.h> /* isnan() */
4
5 redisSortOperation *createSortOperation(int type, robj *pattern) {
6 redisSortOperation *so = zmalloc(sizeof(*so));
7 so->type = type;
8 so->pattern = pattern;
9 return so;
10 }
11
12 /* Return the value associated to the key with a name obtained using
13 * the following rules:
14 *
15 * 1) The first occurence of '*' in 'pattern' is substituted with 'subst'.
16 *
17 * 2) If 'pattern' matches the "->" string, everything on the left of
18 * the arrow is treated as the name of an hash field, and the part on the
19 * left as the key name containing an hash. The value of the specified
20 * field is returned.
21 *
22 * 3) If 'pattern' equals "#", the function simply returns 'subst' itself so
23 * that the SORT command can be used like: SORT key GET # to retrieve
24 * the Set/List elements directly.
25 *
26 * The returned object will always have its refcount increased by 1
27 * when it is non-NULL. */
28 robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
29 char *p, *f, *k;
30 sds spat, ssub;
31 robj *keyobj, *fieldobj = NULL, *o;
32 int prefixlen, sublen, postfixlen, fieldlen;
33
34 /* If the pattern is "#" return the substitution object itself in order
35 * to implement the "SORT ... GET #" feature. */
36 spat = pattern->ptr;
37 if (spat[0] == '#' && spat[1] == '\0') {
38 incrRefCount(subst);
39 return subst;
40 }
41
42 /* The substitution object may be specially encoded. If so we create
43 * a decoded object on the fly. Otherwise getDecodedObject will just
44 * increment the ref count, that we'll decrement later. */
45 subst = getDecodedObject(subst);
46 ssub = subst->ptr;
47
48 /* If we can't find '*' in the pattern we return NULL as to GET a
49 * fixed key does not make sense. */
50 p = strchr(spat,'*');
51 if (!p) {
52 decrRefCount(subst);
53 return NULL;
54 }
55
56 /* Find out if we're dealing with a hash dereference. */
57 if ((f = strstr(p+1, "->")) != NULL && *(f+2) != '\0') {
58 fieldlen = sdslen(spat)-(f-spat)-2;
59 fieldobj = createStringObject(f+2,fieldlen);
60 } else {
61 fieldlen = 0;
62 }
63
64 /* Perform the '*' substitution. */
65 prefixlen = p-spat;
66 sublen = sdslen(ssub);
67 postfixlen = sdslen(spat)-(prefixlen+1)-(fieldlen ? fieldlen+2 : 0);
68 keyobj = createStringObject(NULL,prefixlen+sublen+postfixlen);
69 k = keyobj->ptr;
70 memcpy(k,spat,prefixlen);
71 memcpy(k+prefixlen,ssub,sublen);
72 memcpy(k+prefixlen+sublen,p+1,postfixlen);
73 decrRefCount(subst); /* Incremented by decodeObject() */
74
75 /* Lookup substituted key */
76 o = lookupKeyRead(db,keyobj);
77 if (o == NULL) goto noobj;
78
79 if (fieldobj) {
80 if (o->type != REDIS_HASH) goto noobj;
81
82 /* Retrieve value from hash by the field name. This operation
83 * already increases the refcount of the returned object. */
84 o = hashTypeGetObject(o, fieldobj);
85 } else {
86 if (o->type != REDIS_STRING) goto noobj;
87
88 /* Every object that this function returns needs to have its refcount
89 * increased. sortCommand decreases it again. */
90 incrRefCount(o);
91 }
92 decrRefCount(keyobj);
93 if (fieldobj) decrRefCount(fieldobj);
94 return o;
95
96 noobj:
97 decrRefCount(keyobj);
98 if (fieldlen) decrRefCount(fieldobj);
99 return NULL;
100 }
101
102 /* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
103 * the additional parameter is not standard but a BSD-specific we have to
104 * pass sorting parameters via the global 'server' structure */
105 int sortCompare(const void *s1, const void *s2) {
106 const redisSortObject *so1 = s1, *so2 = s2;
107 int cmp;
108
109 if (!server.sort_alpha) {
110 /* Numeric sorting. Here it's trivial as we precomputed scores */
111 if (so1->u.score > so2->u.score) {
112 cmp = 1;
113 } else if (so1->u.score < so2->u.score) {
114 cmp = -1;
115 } else {
116 /* Objects have the same score, but we don't want the comparison
117 * to be undefined, so we compare objects lexicographycally.
118 * This way the result of SORT is deterministic. */
119 cmp = compareStringObjects(so1->obj,so2->obj);
120 }
121 } else {
122 /* Alphanumeric sorting */
123 if (server.sort_bypattern) {
124 if (!so1->u.cmpobj || !so2->u.cmpobj) {
125 /* At least one compare object is NULL */
126 if (so1->u.cmpobj == so2->u.cmpobj)
127 cmp = 0;
128 else if (so1->u.cmpobj == NULL)
129 cmp = -1;
130 else
131 cmp = 1;
132 } else {
133 /* We have both the objects, use strcoll */
134 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
135 }
136 } else {
137 /* Compare elements directly. */
138 cmp = compareStringObjects(so1->obj,so2->obj);
139 }
140 }
141 return server.sort_desc ? -cmp : cmp;
142 }
143
144 /* The SORT command is the most complex command in Redis. Warning: this code
145 * is optimized for speed and a bit less for readability */
146 void sortCommand(redisClient *c) {
147 list *operations;
148 unsigned int outputlen = 0;
149 int desc = 0, alpha = 0;
150 long limit_start = 0, limit_count = -1, start, end;
151 int j, dontsort = 0, vectorlen;
152 int getop = 0; /* GET operation counter */
153 int int_convertion_error = 0;
154 robj *sortval, *sortby = NULL, *storekey = NULL;
155 redisSortObject *vector; /* Resulting vector to sort */
156
157 /* Lookup the key to sort. It must be of the right types */
158 sortval = lookupKeyRead(c->db,c->argv[1]);
159 if (sortval && sortval->type != REDIS_SET && sortval->type != REDIS_LIST &&
160 sortval->type != REDIS_ZSET)
161 {
162 addReply(c,shared.wrongtypeerr);
163 return;
164 }
165
166 /* Create a list of operations to perform for every sorted element.
167 * Operations can be GET/DEL/INCR/DECR */
168 operations = listCreate();
169 listSetFreeMethod(operations,zfree);
170 j = 2;
171
172 /* Now we need to protect sortval incrementing its count, in the future
173 * SORT may have options able to overwrite/delete keys during the sorting
174 * and the sorted key itself may get destroied */
175 if (sortval)
176 incrRefCount(sortval);
177 else
178 sortval = createListObject();
179
180 /* The SORT command has an SQL-alike syntax, parse it */
181 while(j < c->argc) {
182 int leftargs = c->argc-j-1;
183 if (!strcasecmp(c->argv[j]->ptr,"asc")) {
184 desc = 0;
185 } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
186 desc = 1;
187 } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
188 alpha = 1;
189 } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
190 if ((getLongFromObjectOrReply(c, c->argv[j+1], &limit_start, NULL) != REDIS_OK) ||
191 (getLongFromObjectOrReply(c, c->argv[j+2], &limit_count, NULL) != REDIS_OK)) return;
192 j+=2;
193 } else if (!strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
194 storekey = c->argv[j+1];
195 j++;
196 } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
197 sortby = c->argv[j+1];
198 /* If the BY pattern does not contain '*', i.e. it is constant,
199 * we don't need to sort nor to lookup the weight keys. */
200 if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
201 j++;
202 } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
203 listAddNodeTail(operations,createSortOperation(
204 REDIS_SORT_GET,c->argv[j+1]));
205 getop++;
206 j++;
207 } else {
208 decrRefCount(sortval);
209 listRelease(operations);
210 addReply(c,shared.syntaxerr);
211 return;
212 }
213 j++;
214 }
215
216 /* If we have STORE we need to force sorting for deterministic output
217 * and replication. We use alpha sorting since this is guaranteed to
218 * work with any input.
219 *
220 * We also want determinism when SORT is called from Lua scripts, so
221 * in this case we also force alpha sorting. */
222 if ((storekey || c->flags & REDIS_LUA_CLIENT) && dontsort) {
223 dontsort = 0;
224 alpha = 1;
225 sortby = NULL;
226 }
227
228 /* Destructively convert encoded sorted sets for SORT. */
229 if (sortval->type == REDIS_ZSET)
230 zsetConvert(sortval, REDIS_ENCODING_SKIPLIST);
231
232 /* Load the sorting vector with all the objects to sort */
233 switch(sortval->type) {
234 case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
235 case REDIS_SET: vectorlen = setTypeSize(sortval); break;
236 case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
237 default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
238 }
239 vector = zmalloc(sizeof(redisSortObject)*vectorlen);
240 j = 0;
241
242 if (sortval->type == REDIS_LIST) {
243 listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
244 listTypeEntry entry;
245 while(listTypeNext(li,&entry)) {
246 vector[j].obj = listTypeGet(&entry);
247 vector[j].u.score = 0;
248 vector[j].u.cmpobj = NULL;
249 j++;
250 }
251 listTypeReleaseIterator(li);
252 } else if (sortval->type == REDIS_SET) {
253 setTypeIterator *si = setTypeInitIterator(sortval);
254 robj *ele;
255 while((ele = setTypeNextObject(si)) != NULL) {
256 vector[j].obj = ele;
257 vector[j].u.score = 0;
258 vector[j].u.cmpobj = NULL;
259 j++;
260 }
261 setTypeReleaseIterator(si);
262 } else if (sortval->type == REDIS_ZSET) {
263 dict *set = ((zset*)sortval->ptr)->dict;
264 dictIterator *di;
265 dictEntry *setele;
266 di = dictGetIterator(set);
267 while((setele = dictNext(di)) != NULL) {
268 vector[j].obj = dictGetKey(setele);
269 vector[j].u.score = 0;
270 vector[j].u.cmpobj = NULL;
271 j++;
272 }
273 dictReleaseIterator(di);
274 } else {
275 redisPanic("Unknown type");
276 }
277 redisAssertWithInfo(c,sortval,j == vectorlen);
278
279 /* Now it's time to load the right scores in the sorting vector */
280 if (dontsort == 0) {
281 for (j = 0; j < vectorlen; j++) {
282 robj *byval;
283 if (sortby) {
284 /* lookup value to sort by */
285 byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
286 if (!byval) continue;
287 } else {
288 /* use object itself to sort by */
289 byval = vector[j].obj;
290 }
291
292 if (alpha) {
293 if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
294 } else {
295 if (byval->encoding == REDIS_ENCODING_RAW) {
296 char *eptr;
297
298 vector[j].u.score = strtod(byval->ptr,&eptr);
299 if (eptr[0] != '\0' || errno == ERANGE ||
300 isnan(vector[j].u.score))
301 {
302 int_convertion_error = 1;
303 }
304 } else if (byval->encoding == REDIS_ENCODING_INT) {
305 /* Don't need to decode the object if it's
306 * integer-encoded (the only encoding supported) so
307 * far. We can just cast it */
308 vector[j].u.score = (long)byval->ptr;
309 } else {
310 redisAssertWithInfo(c,sortval,1 != 1);
311 }
312 }
313
314 /* when the object was retrieved using lookupKeyByPattern,
315 * its refcount needs to be decreased. */
316 if (sortby) {
317 decrRefCount(byval);
318 }
319 }
320 }
321
322 /* We are ready to sort the vector... perform a bit of sanity check
323 * on the LIMIT option too. We'll use a partial version of quicksort. */
324 start = (limit_start < 0) ? 0 : limit_start;
325 end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
326 if (start >= vectorlen) {
327 start = vectorlen-1;
328 end = vectorlen-2;
329 }
330 if (end >= vectorlen) end = vectorlen-1;
331
332 if (dontsort == 0) {
333 server.sort_desc = desc;
334 server.sort_alpha = alpha;
335 server.sort_bypattern = sortby ? 1 : 0;
336 if (sortby && (start != 0 || end != vectorlen-1))
337 pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
338 else
339 qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
340 }
341
342 /* Send command output to the output buffer, performing the specified
343 * GET/DEL/INCR/DECR operations if any. */
344 outputlen = getop ? getop*(end-start+1) : end-start+1;
345 if (int_convertion_error) {
346 addReplyError(c,"One or more scores can't be converted into double");
347 } else if (storekey == NULL) {
348 /* STORE option not specified, sent the sorting result to client */
349 addReplyMultiBulkLen(c,outputlen);
350 for (j = start; j <= end; j++) {
351 listNode *ln;
352 listIter li;
353
354 if (!getop) addReplyBulk(c,vector[j].obj);
355 listRewind(operations,&li);
356 while((ln = listNext(&li))) {
357 redisSortOperation *sop = ln->value;
358 robj *val = lookupKeyByPattern(c->db,sop->pattern,
359 vector[j].obj);
360
361 if (sop->type == REDIS_SORT_GET) {
362 if (!val) {
363 addReply(c,shared.nullbulk);
364 } else {
365 addReplyBulk(c,val);
366 decrRefCount(val);
367 }
368 } else {
369 /* Always fails */
370 redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
371 }
372 }
373 }
374 } else {
375 robj *sobj = createZiplistObject();
376
377 /* STORE option specified, set the sorting result as a List object */
378 for (j = start; j <= end; j++) {
379 listNode *ln;
380 listIter li;
381
382 if (!getop) {
383 listTypePush(sobj,vector[j].obj,REDIS_TAIL);
384 } else {
385 listRewind(operations,&li);
386 while((ln = listNext(&li))) {
387 redisSortOperation *sop = ln->value;
388 robj *val = lookupKeyByPattern(c->db,sop->pattern,
389 vector[j].obj);
390
391 if (sop->type == REDIS_SORT_GET) {
392 if (!val) val = createStringObject("",0);
393
394 /* listTypePush does an incrRefCount, so we should take care
395 * care of the incremented refcount caused by either
396 * lookupKeyByPattern or createStringObject("",0) */
397 listTypePush(sobj,val,REDIS_TAIL);
398 decrRefCount(val);
399 } else {
400 /* Always fails */
401 redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
402 }
403 }
404 }
405 }
406 if (outputlen) {
407 setKey(c->db,storekey,sobj);
408 server.dirty += outputlen;
409 } else if (dbDelete(c->db,storekey)) {
410 signalModifiedKey(c->db,storekey);
411 server.dirty++;
412 }
413 decrRefCount(sobj);
414 addReplyLongLong(c,outputlen);
415 }
416
417 /* Cleanup */
418 if (sortval->type == REDIS_LIST || sortval->type == REDIS_SET)
419 for (j = 0; j < vectorlen; j++)
420 decrRefCount(vector[j].obj);
421 decrRefCount(sortval);
422 listRelease(operations);
423 for (j = 0; j < vectorlen; j++) {
424 if (alpha && vector[j].u.cmpobj)
425 decrRefCount(vector[j].u.cmpobj);
426 }
427 zfree(vector);
428 }
429
430