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