src/sort.c

#include "redis.h"
#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
#include <math.h> /* isnan() */

redisSortOperation *createSortOperation(int type, robj *pattern) {
    redisSortOperation *so = zmalloc(sizeof(*so));
    so->type = type;
    so->pattern = pattern;
    return so;
}

/* Return the value associated to the key with a name obtained using
 * the following rules:
 *
 * 1) The first occurence of '*' in 'pattern' is substituted with 'subst'.
 *
 * 2) If 'pattern' matches the "->" string, everything on the left of
 *    the arrow is treated as the name of an hash field, and the part on the
 *    left as the key name containing an hash. The value of the specified
 *    field is returned.
 *
 * 3) If 'pattern' equals "#", the function simply returns 'subst' itself so
 *    that the SORT command can be used like: SORT key GET # to retrieve
 *    the Set/List elements directly.
 *
 * The returned object will always have its refcount increased by 1
 * when it is non-NULL. */
robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
    char *p, *f, *k;
    sds spat, ssub;
    robj *keyobj, *fieldobj = NULL, *o;
    int prefixlen, sublen, postfixlen, fieldlen;

    /* If the pattern is "#" return the substitution object itself in order
     * to implement the "SORT ... GET #" feature. */
    spat = pattern->ptr;
    if (spat[0] == '#' && spat[1] == '\0') {
        incrRefCount(subst);
        return subst;
    }

    /* The substitution object may be specially encoded. If so we create
     * a decoded object on the fly. Otherwise getDecodedObject will just
     * increment the ref count, that we'll decrement later. */
    subst = getDecodedObject(subst);
    ssub = subst->ptr;

    /* If we can't find '*' in the pattern we return NULL as to GET a
     * fixed key does not make sense. */
    p = strchr(spat,'*');
    if (!p) {
        decrRefCount(subst);
        return NULL;
    }

    /* Find out if we're dealing with a hash dereference. */
    if ((f = strstr(p+1, "->")) != NULL && *(f+2) != '\0') {
        fieldlen = sdslen(spat)-(f-spat)-2;
        fieldobj = createStringObject(f+2,fieldlen);
    } else {
        fieldlen = 0;
    }

    /* Perform the '*' substitution. */
    prefixlen = p-spat;
    sublen = sdslen(ssub);
    postfixlen = sdslen(spat)-(prefixlen+1)-(fieldlen ? fieldlen+2 : 0);
    keyobj = createStringObject(NULL,prefixlen+sublen+postfixlen);
    k = keyobj->ptr;
    memcpy(k,spat,prefixlen);
    memcpy(k+prefixlen,ssub,sublen);
    memcpy(k+prefixlen+sublen,p+1,postfixlen);
    decrRefCount(subst); /* Incremented by decodeObject() */

    /* Lookup substituted key */
    o = lookupKeyRead(db,keyobj);
    if (o == NULL) goto noobj;

    if (fieldobj) {
        if (o->type != REDIS_HASH) goto noobj;

        /* Retrieve value from hash by the field name. This operation
         * already increases the refcount of the returned object. */
        o = hashTypeGetObject(o, fieldobj);
    } else {
        if (o->type != REDIS_STRING) goto noobj;

        /* Every object that this function returns needs to have its refcount
         * increased. sortCommand decreases it again. */
        incrRefCount(o);
    }
    decrRefCount(keyobj);
    if (fieldobj) decrRefCount(fieldobj);
    return o;

noobj:
    decrRefCount(keyobj);
    if (fieldlen) decrRefCount(fieldobj);
    return NULL;
}

/* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
 * the additional parameter is not standard but a BSD-specific we have to
 * pass sorting parameters via the global 'server' structure */
int sortCompare(const void *s1, const void *s2) {
    const redisSortObject *so1 = s1, *so2 = s2;
    int cmp;

    if (!server.sort_alpha) {
        /* Numeric sorting. Here it's trivial as we precomputed scores */
        if (so1->u.score > so2->u.score) {
            cmp = 1;
        } else if (so1->u.score < so2->u.score) {
            cmp = -1;
        } else {
            /* Objects have the same score, but we don't want the comparison
             * to be undefined, so we compare objects lexicographycally.
             * This way the result of SORT is deterministic. */
            cmp = compareStringObjects(so1->obj,so2->obj);
        }
    } else {
        /* Alphanumeric sorting */
        if (server.sort_bypattern) {
            if (!so1->u.cmpobj || !so2->u.cmpobj) {
                /* At least one compare object is NULL */
                if (so1->u.cmpobj == so2->u.cmpobj)
                    cmp = 0;
                else if (so1->u.cmpobj == NULL)
                    cmp = -1;
                else
                    cmp = 1;
            } else {
                /* We have both the objects, use strcoll */
                cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
            }
        } else {
            /* Compare elements directly. */
            cmp = compareStringObjects(so1->obj,so2->obj);
        }
    }
    return server.sort_desc ? -cmp : cmp;
}

/* The SORT command is the most complex command in Redis. Warning: this code
 * is optimized for speed and a bit less for readability */
void sortCommand(redisClient *c) {
    list *operations;
    unsigned int outputlen = 0;
    int desc = 0, alpha = 0;
    long limit_start = 0, limit_count = -1, start, end;
    int j, dontsort = 0, vectorlen;
    int getop = 0; /* GET operation counter */
    int int_convertion_error = 0;
    robj *sortval, *sortby = NULL, *storekey = NULL;
    redisSortObject *vector; /* Resulting vector to sort */

    /* Lookup the key to sort. It must be of the right types */
    sortval = lookupKeyRead(c->db,c->argv[1]);
    if (sortval && sortval->type != REDIS_SET && sortval->type != REDIS_LIST &&
        sortval->type != REDIS_ZSET)
    {
        addReply(c,shared.wrongtypeerr);
        return;
    }

    /* Create a list of operations to perform for every sorted element.
     * Operations can be GET/DEL/INCR/DECR */
    operations = listCreate();
    listSetFreeMethod(operations,zfree);
    j = 2;

    /* Now we need to protect sortval incrementing its count, in the future
     * SORT may have options able to overwrite/delete keys during the sorting
     * and the sorted key itself may get destroied */
    if (sortval)
        incrRefCount(sortval);
    else
        sortval = createListObject();

    /* The SORT command has an SQL-alike syntax, parse it */
    while(j < c->argc) {
        int leftargs = c->argc-j-1;
        if (!strcasecmp(c->argv[j]->ptr,"asc")) {
            desc = 0;
        } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
            desc = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
            alpha = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
            if ((getLongFromObjectOrReply(c, c->argv[j+1], &limit_start, NULL) != REDIS_OK) ||
                (getLongFromObjectOrReply(c, c->argv[j+2], &limit_count, NULL) != REDIS_OK)) return;
            j+=2;
        } else if (!strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
            storekey = c->argv[j+1];
            j++;
        } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
            sortby = c->argv[j+1];
            /* If the BY pattern does not contain '*', i.e. it is constant,
             * we don't need to sort nor to lookup the weight keys. */
            if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
            j++;
        } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
            listAddNodeTail(operations,createSortOperation(
                REDIS_SORT_GET,c->argv[j+1]));
            getop++;
            j++;
        } else {
            decrRefCount(sortval);
            listRelease(operations);
            addReply(c,shared.syntaxerr);
            return;
        }
        j++;
    }

    /* If we have STORE we need to force sorting for deterministic output
     * and replication. We use alpha sorting since this is guaranteed to
     * work with any input. */
    if (storekey && dontsort) {
        dontsort = 0;
        alpha = 1;
        sortby = NULL;
    }

    /* Destructively convert encoded sorted sets for SORT. */
    if (sortval->type == REDIS_ZSET)
        zsetConvert(sortval, REDIS_ENCODING_SKIPLIST);

    /* Load the sorting vector with all the objects to sort */
    switch(sortval->type) {
    case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
    case REDIS_SET: vectorlen =  setTypeSize(sortval); break;
    case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
    default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
    }
    vector = zmalloc(sizeof(redisSortObject)*vectorlen);
    j = 0;

    if (sortval->type == REDIS_LIST) {
        listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
        listTypeEntry entry;
        while(listTypeNext(li,&entry)) {
            vector[j].obj = listTypeGet(&entry);
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        listTypeReleaseIterator(li);
    } else if (sortval->type == REDIS_SET) {
        setTypeIterator *si = setTypeInitIterator(sortval);
        robj *ele;
        while((ele = setTypeNextObject(si)) != NULL) {
            vector[j].obj = ele;
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        setTypeReleaseIterator(si);
    } else if (sortval->type == REDIS_ZSET) {
        dict *set = ((zset*)sortval->ptr)->dict;
        dictIterator *di;
        dictEntry *setele;
        di = dictGetIterator(set);
        while((setele = dictNext(di)) != NULL) {
            vector[j].obj = dictGetKey(setele);
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        dictReleaseIterator(di);
    } else {
        redisPanic("Unknown type");
    }
    redisAssertWithInfo(c,sortval,j == vectorlen);

    /* Now it's time to load the right scores in the sorting vector */
    if (dontsort == 0) {
        for (j = 0; j < vectorlen; j++) {
            robj *byval;
            if (sortby) {
                /* lookup value to sort by */
                byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
                if (!byval) continue;
            } else {
                /* use object itself to sort by */
                byval = vector[j].obj;
            }

            if (alpha) {
                if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
            } else {
                if (byval->encoding == REDIS_ENCODING_RAW) {
                    char *eptr;

                    vector[j].u.score = strtod(byval->ptr,&eptr);
                    if (eptr[0] != '\0' || errno == ERANGE ||
                        isnan(vector[j].u.score))
                    {
                        int_convertion_error = 1;
                    }
                } else if (byval->encoding == REDIS_ENCODING_INT) {
                    /* Don't need to decode the object if it's
                     * integer-encoded (the only encoding supported) so
                     * far. We can just cast it */
                    vector[j].u.score = (long)byval->ptr;
                } else {
                    redisAssertWithInfo(c,sortval,1 != 1);
                }
            }

            /* when the object was retrieved using lookupKeyByPattern,
             * its refcount needs to be decreased. */
            if (sortby) {
                decrRefCount(byval);
            }
        }
    }

    /* We are ready to sort the vector... perform a bit of sanity check
     * on the LIMIT option too. We'll use a partial version of quicksort. */
    start = (limit_start < 0) ? 0 : limit_start;
    end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
    if (start >= vectorlen) {
        start = vectorlen-1;
        end = vectorlen-2;
    }
    if (end >= vectorlen) end = vectorlen-1;

    server.sort_dontsort = dontsort;
    if (dontsort == 0) {
        server.sort_desc = desc;
        server.sort_alpha = alpha;
        server.sort_bypattern = sortby ? 1 : 0;
        if (sortby && (start != 0 || end != vectorlen-1))
            pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
        else
            qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
    }

    /* Send command output to the output buffer, performing the specified
     * GET/DEL/INCR/DECR operations if any. */
    outputlen = getop ? getop*(end-start+1) : end-start+1;
    if (int_convertion_error) {
        addReplyError(c,"One or more scores can't be converted into double");
    } else if (storekey == NULL) {
        /* STORE option not specified, sent the sorting result to client */
        addReplyMultiBulkLen(c,outputlen);
        for (j = start; j <= end; j++) {
            listNode *ln;
            listIter li;

            if (!getop) addReplyBulk(c,vector[j].obj);
            listRewind(operations,&li);
            while((ln = listNext(&li))) {
                redisSortOperation *sop = ln->value;
                robj *val = lookupKeyByPattern(c->db,sop->pattern,
                    vector[j].obj);

                if (sop->type == REDIS_SORT_GET) {
                    if (!val) {
                        addReply(c,shared.nullbulk);
                    } else {
                        addReplyBulk(c,val);
                        decrRefCount(val);
                    }
                } else {
                    /* Always fails */
                    redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
                }
            }
        }
    } else {
        robj *sobj = createZiplistObject();

        /* STORE option specified, set the sorting result as a List object */
        for (j = start; j <= end; j++) {
            listNode *ln;
            listIter li;

            if (!getop) {
                listTypePush(sobj,vector[j].obj,REDIS_TAIL);
            } else {
                listRewind(operations,&li);
                while((ln = listNext(&li))) {
                    redisSortOperation *sop = ln->value;
                    robj *val = lookupKeyByPattern(c->db,sop->pattern,
                        vector[j].obj);

                    if (sop->type == REDIS_SORT_GET) {
                        if (!val) val = createStringObject("",0);

                        /* listTypePush does an incrRefCount, so we should take care
                         * care of the incremented refcount caused by either
                         * lookupKeyByPattern or createStringObject("",0) */
                        listTypePush(sobj,val,REDIS_TAIL);
                        decrRefCount(val);
                    } else {
                        /* Always fails */
                        redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
                    }
                }
            }
        }
        if (outputlen) {
            setKey(c->db,storekey,sobj);
            server.dirty += outputlen;
        } else if (dbDelete(c->db,storekey)) {
            signalModifiedKey(c->db,storekey);
            server.dirty++;
        }
        decrRefCount(sobj);
        addReplyLongLong(c,outputlen);
    }

    /* Cleanup */
    if (sortval->type == REDIS_LIST || sortval->type == REDIS_SET)
        for (j = 0; j < vectorlen; j++)
            decrRefCount(vector[j].obj);
    decrRefCount(sortval);
    listRelease(operations);
    for (j = 0; j < vectorlen; j++) {
        if (alpha && vector[j].u.cmpobj)
            decrRefCount(vector[j].u.cmpobj);
    }
    zfree(vector);
}
Commit	Line	Data
	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	if (storekey && dontsort) {
	220	dontsort = 0;
	221	alpha = 1;
	222	sortby = NULL;
	223	}
	224
	225	/* Destructively convert encoded sorted sets for SORT. */
	226	if (sortval->type == REDIS_ZSET)
	227	zsetConvert(sortval, REDIS_ENCODING_SKIPLIST);
	228
	229	/* Load the sorting vector with all the objects to sort */
	230	switch(sortval->type) {
	231	case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
	232	case REDIS_SET: vectorlen = setTypeSize(sortval); break;
	233	case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
	234	default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
	235	}
	236	vector = zmalloc(sizeof(redisSortObject)*vectorlen);
	237	j = 0;
	238
	239	if (sortval->type == REDIS_LIST) {
	240	listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
	241	listTypeEntry entry;
	242	while(listTypeNext(li,&entry)) {
	243	vector[j].obj = listTypeGet(&entry);
	244	vector[j].u.score = 0;
	245	vector[j].u.cmpobj = NULL;
	246	j++;
	247	}
	248	listTypeReleaseIterator(li);
	249	} else if (sortval->type == REDIS_SET) {
	250	setTypeIterator *si = setTypeInitIterator(sortval);
	251	robj *ele;
	252	while((ele = setTypeNextObject(si)) != NULL) {
	253	vector[j].obj = ele;
	254	vector[j].u.score = 0;
	255	vector[j].u.cmpobj = NULL;
	256	j++;
	257	}
	258	setTypeReleaseIterator(si);
	259	} else if (sortval->type == REDIS_ZSET) {
	260	dict set = ((zset)sortval->ptr)->dict;
	261	dictIterator *di;
	262	dictEntry *setele;
	263	di = dictGetIterator(set);
	264	while((setele = dictNext(di)) != NULL) {
	265	vector[j].obj = dictGetKey(setele);
	266	vector[j].u.score = 0;
	267	vector[j].u.cmpobj = NULL;
	268	j++;
	269	}
	270	dictReleaseIterator(di);
	271	} else {
	272	redisPanic("Unknown type");
	273	}
	274	redisAssertWithInfo(c,sortval,j == vectorlen);
	275
	276	/* Now it's time to load the right scores in the sorting vector */
	277	if (dontsort == 0) {
	278	for (j = 0; j < vectorlen; j++) {
	279	robj *byval;
	280	if (sortby) {
	281	/* lookup value to sort by */
	282	byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
	283	if (!byval) continue;
	284	} else {
	285	/* use object itself to sort by */
	286	byval = vector[j].obj;
	287	}
	288
	289	if (alpha) {
	290	if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
	291	} else {
	292	if (byval->encoding == REDIS_ENCODING_RAW) {
	293	char *eptr;
	294
	295	vector[j].u.score = strtod(byval->ptr,&eptr);
	296	if (eptr[0] != '\0' \|\| errno == ERANGE \|\|
	297	isnan(vector[j].u.score))
	298	{
	299	int_convertion_error = 1;
	300	}
	301	} else if (byval->encoding == REDIS_ENCODING_INT) {
	302	/* Don't need to decode the object if it's
	303	* integer-encoded (the only encoding supported) so
	304	* far. We can just cast it */
	305	vector[j].u.score = (long)byval->ptr;
	306	} else {
	307	redisAssertWithInfo(c,sortval,1 != 1);
	308	}
	309	}
	310
	311	/* when the object was retrieved using lookupKeyByPattern,
	312	* its refcount needs to be decreased. */
	313	if (sortby) {
	314	decrRefCount(byval);
	315	}
	316	}
	317	}
	318
	319	/* We are ready to sort the vector... perform a bit of sanity check
	320	* on the LIMIT option too. We'll use a partial version of quicksort. */
	321	start = (limit_start < 0) ? 0 : limit_start;
	322	end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
	323	if (start >= vectorlen) {
	324	start = vectorlen-1;
	325	end = vectorlen-2;
	326	}
	327	if (end >= vectorlen) end = vectorlen-1;
	328
	329	server.sort_dontsort = dontsort;
	330	if (dontsort == 0) {
	331	server.sort_desc = desc;
	332	server.sort_alpha = alpha;
	333	server.sort_bypattern = sortby ? 1 : 0;
	334	if (sortby && (start != 0 \|\| end != vectorlen-1))
	335	pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
	336	else
	337	qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
	338	}
	339
	340	/* Send command output to the output buffer, performing the specified
	341	* GET/DEL/INCR/DECR operations if any. */
	342	outputlen = getop ? getop*(end-start+1) : end-start+1;
	343	if (int_convertion_error) {
	344	addReplyError(c,"One or more scores can't be converted into double");
	345	} else if (storekey == NULL) {
	346	/* STORE option not specified, sent the sorting result to client */
	347	addReplyMultiBulkLen(c,outputlen);
	348	for (j = start; j <= end; j++) {
	349	listNode *ln;
	350	listIter li;
	351
	352	if (!getop) addReplyBulk(c,vector[j].obj);
	353	listRewind(operations,&li);
	354	while((ln = listNext(&li))) {
	355	redisSortOperation *sop = ln->value;
	356	robj *val = lookupKeyByPattern(c->db,sop->pattern,
	357	vector[j].obj);
	358
	359	if (sop->type == REDIS_SORT_GET) {
	360	if (!val) {
	361	addReply(c,shared.nullbulk);
	362	} else {
	363	addReplyBulk(c,val);
	364	decrRefCount(val);
	365	}
	366	} else {
	367	/* Always fails */
	368	redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
	369	}
	370	}
	371	}
	372	} else {
	373	robj *sobj = createZiplistObject();
	374
	375	/* STORE option specified, set the sorting result as a List object */
	376	for (j = start; j <= end; j++) {
	377	listNode *ln;
	378	listIter li;
	379
	380	if (!getop) {
	381	listTypePush(sobj,vector[j].obj,REDIS_TAIL);
	382	} else {
	383	listRewind(operations,&li);
	384	while((ln = listNext(&li))) {
	385	redisSortOperation *sop = ln->value;
	386	robj *val = lookupKeyByPattern(c->db,sop->pattern,
	387	vector[j].obj);
	388
	389	if (sop->type == REDIS_SORT_GET) {
	390	if (!val) val = createStringObject("",0);
	391
	392	/* listTypePush does an incrRefCount, so we should take care
	393	* care of the incremented refcount caused by either
	394	* lookupKeyByPattern or createStringObject("",0) */
	395	listTypePush(sobj,val,REDIS_TAIL);
	396	decrRefCount(val);
	397	} else {
	398	/* Always fails */
	399	redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
	400	}
	401	}
	402	}
	403	}
	404	if (outputlen) {
	405	setKey(c->db,storekey,sobj);
	406	server.dirty += outputlen;
	407	} else if (dbDelete(c->db,storekey)) {
	408	signalModifiedKey(c->db,storekey);
	409	server.dirty++;
	410	}
	411	decrRefCount(sobj);
	412	addReplyLongLong(c,outputlen);
	413	}
	414
	415	/* Cleanup */
	416	if (sortval->type == REDIS_LIST \|\| sortval->type == REDIS_SET)
	417	for (j = 0; j < vectorlen; j++)
	418	decrRefCount(vector[j].obj);
	419	decrRefCount(sortval);
	420	listRelease(operations);
	421	for (j = 0; j < vectorlen; j++) {
	422	if (alpha && vector[j].u.cmpobj)
	423	decrRefCount(vector[j].u.cmpobj);
	424	}
	425	zfree(vector);
	426	}
	427
	428