]> git.saurik.com Git - redis.git/blob - src/t_set.c
Update target encoding for sorted set from rdb
[redis.git] / src / t_set.c
1 #include "redis.h"
2
3 /*-----------------------------------------------------------------------------
4 * Set Commands
5 *----------------------------------------------------------------------------*/
6
7 /* Factory method to return a set that *can* hold "value". When the object has
8 * an integer-encodable value, an intset will be returned. Otherwise a regular
9 * hash table. */
10 robj *setTypeCreate(robj *value) {
11 if (isObjectRepresentableAsLongLong(value,NULL) == REDIS_OK)
12 return createIntsetObject();
13 return createSetObject();
14 }
15
16 int setTypeAdd(robj *subject, robj *value) {
17 long long llval;
18 if (subject->encoding == REDIS_ENCODING_HT) {
19 if (dictAdd(subject->ptr,value,NULL) == DICT_OK) {
20 incrRefCount(value);
21 return 1;
22 }
23 } else if (subject->encoding == REDIS_ENCODING_INTSET) {
24 if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
25 uint8_t success = 0;
26 subject->ptr = intsetAdd(subject->ptr,llval,&success);
27 if (success) {
28 /* Convert to regular set when the intset contains
29 * too many entries. */
30 if (intsetLen(subject->ptr) > server.set_max_intset_entries)
31 setTypeConvert(subject,REDIS_ENCODING_HT);
32 return 1;
33 }
34 } else {
35 /* Failed to get integer from object, convert to regular set. */
36 setTypeConvert(subject,REDIS_ENCODING_HT);
37
38 /* The set *was* an intset and this value is not integer
39 * encodable, so dictAdd should always work. */
40 redisAssert(dictAdd(subject->ptr,value,NULL) == DICT_OK);
41 incrRefCount(value);
42 return 1;
43 }
44 } else {
45 redisPanic("Unknown set encoding");
46 }
47 return 0;
48 }
49
50 int setTypeRemove(robj *setobj, robj *value) {
51 long long llval;
52 if (setobj->encoding == REDIS_ENCODING_HT) {
53 if (dictDelete(setobj->ptr,value) == DICT_OK) {
54 if (htNeedsResize(setobj->ptr)) dictResize(setobj->ptr);
55 return 1;
56 }
57 } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
58 if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
59 int success;
60 setobj->ptr = intsetRemove(setobj->ptr,llval,&success);
61 if (success) return 1;
62 }
63 } else {
64 redisPanic("Unknown set encoding");
65 }
66 return 0;
67 }
68
69 int setTypeIsMember(robj *subject, robj *value) {
70 long long llval;
71 if (subject->encoding == REDIS_ENCODING_HT) {
72 return dictFind((dict*)subject->ptr,value) != NULL;
73 } else if (subject->encoding == REDIS_ENCODING_INTSET) {
74 if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
75 return intsetFind((intset*)subject->ptr,llval);
76 }
77 } else {
78 redisPanic("Unknown set encoding");
79 }
80 return 0;
81 }
82
83 setTypeIterator *setTypeInitIterator(robj *subject) {
84 setTypeIterator *si = zmalloc(sizeof(setTypeIterator));
85 si->subject = subject;
86 si->encoding = subject->encoding;
87 if (si->encoding == REDIS_ENCODING_HT) {
88 si->di = dictGetIterator(subject->ptr);
89 } else if (si->encoding == REDIS_ENCODING_INTSET) {
90 si->ii = 0;
91 } else {
92 redisPanic("Unknown set encoding");
93 }
94 return si;
95 }
96
97 void setTypeReleaseIterator(setTypeIterator *si) {
98 if (si->encoding == REDIS_ENCODING_HT)
99 dictReleaseIterator(si->di);
100 zfree(si);
101 }
102
103 /* Move to the next entry in the set. Returns the object at the current
104 * position.
105 *
106 * Since set elements can be internally be stored as redis objects or
107 * simple arrays of integers, setTypeNext returns the encoding of the
108 * set object you are iterating, and will populate the appropriate pointer
109 * (eobj) or (llobj) accordingly.
110 *
111 * When there are no longer elements -1 is returned.
112 * Returned objects ref count is not incremented, so this function is
113 * copy on write friendly. */
114 int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
115 if (si->encoding == REDIS_ENCODING_HT) {
116 dictEntry *de = dictNext(si->di);
117 if (de == NULL) return -1;
118 *objele = dictGetEntryKey(de);
119 } else if (si->encoding == REDIS_ENCODING_INTSET) {
120 if (!intsetGet(si->subject->ptr,si->ii++,llele))
121 return -1;
122 }
123 return si->encoding;
124 }
125
126 /* The not copy on write friendly version but easy to use version
127 * of setTypeNext() is setTypeNextObject(), returning new objects
128 * or incrementing the ref count of returned objects. So if you don't
129 * retain a pointer to this object you should call decrRefCount() against it.
130 *
131 * This function is the way to go for write operations where COW is not
132 * an issue as the result will be anyway of incrementing the ref count. */
133 robj *setTypeNextObject(setTypeIterator *si) {
134 int64_t intele;
135 robj *objele;
136 int encoding;
137
138 encoding = setTypeNext(si,&objele,&intele);
139 switch(encoding) {
140 case -1: return NULL;
141 case REDIS_ENCODING_INTSET:
142 return createStringObjectFromLongLong(intele);
143 case REDIS_ENCODING_HT:
144 incrRefCount(objele);
145 return objele;
146 default:
147 redisPanic("Unsupported encoding");
148 }
149 return NULL; /* just to suppress warnings */
150 }
151
152 /* Return random element from a non empty set.
153 * The returned element can be a int64_t value if the set is encoded
154 * as an "intset" blob of integers, or a redis object if the set
155 * is a regular set.
156 *
157 * The caller provides both pointers to be populated with the right
158 * object. The return value of the function is the object->encoding
159 * field of the object and is used by the caller to check if the
160 * int64_t pointer or the redis object pointere was populated.
161 *
162 * When an object is returned (the set was a real set) the ref count
163 * of the object is not incremented so this function can be considered
164 * copy on write friendly. */
165 int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele) {
166 if (setobj->encoding == REDIS_ENCODING_HT) {
167 dictEntry *de = dictGetRandomKey(setobj->ptr);
168 *objele = dictGetEntryKey(de);
169 } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
170 *llele = intsetRandom(setobj->ptr);
171 } else {
172 redisPanic("Unknown set encoding");
173 }
174 return setobj->encoding;
175 }
176
177 unsigned long setTypeSize(robj *subject) {
178 if (subject->encoding == REDIS_ENCODING_HT) {
179 return dictSize((dict*)subject->ptr);
180 } else if (subject->encoding == REDIS_ENCODING_INTSET) {
181 return intsetLen((intset*)subject->ptr);
182 } else {
183 redisPanic("Unknown set encoding");
184 }
185 }
186
187 /* Convert the set to specified encoding. The resulting dict (when converting
188 * to a hashtable) is presized to hold the number of elements in the original
189 * set. */
190 void setTypeConvert(robj *setobj, int enc) {
191 setTypeIterator *si;
192 redisAssert(setobj->type == REDIS_SET &&
193 setobj->encoding == REDIS_ENCODING_INTSET);
194
195 if (enc == REDIS_ENCODING_HT) {
196 int64_t intele;
197 dict *d = dictCreate(&setDictType,NULL);
198 robj *element;
199
200 /* Presize the dict to avoid rehashing */
201 dictExpand(d,intsetLen(setobj->ptr));
202
203 /* To add the elements we extract integers and create redis objects */
204 si = setTypeInitIterator(setobj);
205 while (setTypeNext(si,NULL,&intele) != -1) {
206 element = createStringObjectFromLongLong(intele);
207 redisAssert(dictAdd(d,element,NULL) == DICT_OK);
208 }
209 setTypeReleaseIterator(si);
210
211 setobj->encoding = REDIS_ENCODING_HT;
212 zfree(setobj->ptr);
213 setobj->ptr = d;
214 } else {
215 redisPanic("Unsupported set conversion");
216 }
217 }
218
219 void saddCommand(redisClient *c) {
220 robj *set;
221
222 set = lookupKeyWrite(c->db,c->argv[1]);
223 c->argv[2] = tryObjectEncoding(c->argv[2]);
224 if (set == NULL) {
225 set = setTypeCreate(c->argv[2]);
226 dbAdd(c->db,c->argv[1],set);
227 } else {
228 if (set->type != REDIS_SET) {
229 addReply(c,shared.wrongtypeerr);
230 return;
231 }
232 }
233 if (setTypeAdd(set,c->argv[2])) {
234 signalModifiedKey(c->db,c->argv[1]);
235 server.dirty++;
236 addReply(c,shared.cone);
237 } else {
238 addReply(c,shared.czero);
239 }
240 }
241
242 void sremCommand(redisClient *c) {
243 robj *set;
244
245 if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
246 checkType(c,set,REDIS_SET)) return;
247
248 c->argv[2] = tryObjectEncoding(c->argv[2]);
249 if (setTypeRemove(set,c->argv[2])) {
250 if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
251 signalModifiedKey(c->db,c->argv[1]);
252 server.dirty++;
253 addReply(c,shared.cone);
254 } else {
255 addReply(c,shared.czero);
256 }
257 }
258
259 void smoveCommand(redisClient *c) {
260 robj *srcset, *dstset, *ele;
261 srcset = lookupKeyWrite(c->db,c->argv[1]);
262 dstset = lookupKeyWrite(c->db,c->argv[2]);
263 ele = c->argv[3] = tryObjectEncoding(c->argv[3]);
264
265 /* If the source key does not exist return 0 */
266 if (srcset == NULL) {
267 addReply(c,shared.czero);
268 return;
269 }
270
271 /* If the source key has the wrong type, or the destination key
272 * is set and has the wrong type, return with an error. */
273 if (checkType(c,srcset,REDIS_SET) ||
274 (dstset && checkType(c,dstset,REDIS_SET))) return;
275
276 /* If srcset and dstset are equal, SMOVE is a no-op */
277 if (srcset == dstset) {
278 addReply(c,shared.cone);
279 return;
280 }
281
282 /* If the element cannot be removed from the src set, return 0. */
283 if (!setTypeRemove(srcset,ele)) {
284 addReply(c,shared.czero);
285 return;
286 }
287
288 /* Remove the src set from the database when empty */
289 if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);
290 signalModifiedKey(c->db,c->argv[1]);
291 signalModifiedKey(c->db,c->argv[2]);
292 server.dirty++;
293
294 /* Create the destination set when it doesn't exist */
295 if (!dstset) {
296 dstset = setTypeCreate(ele);
297 dbAdd(c->db,c->argv[2],dstset);
298 }
299
300 /* An extra key has changed when ele was successfully added to dstset */
301 if (setTypeAdd(dstset,ele)) server.dirty++;
302 addReply(c,shared.cone);
303 }
304
305 void sismemberCommand(redisClient *c) {
306 robj *set;
307
308 if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
309 checkType(c,set,REDIS_SET)) return;
310
311 c->argv[2] = tryObjectEncoding(c->argv[2]);
312 if (setTypeIsMember(set,c->argv[2]))
313 addReply(c,shared.cone);
314 else
315 addReply(c,shared.czero);
316 }
317
318 void scardCommand(redisClient *c) {
319 robj *o;
320
321 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
322 checkType(c,o,REDIS_SET)) return;
323
324 addReplyLongLong(c,setTypeSize(o));
325 }
326
327 void spopCommand(redisClient *c) {
328 robj *set, *ele;
329 int64_t llele;
330 int encoding;
331
332 if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
333 checkType(c,set,REDIS_SET)) return;
334
335 encoding = setTypeRandomElement(set,&ele,&llele);
336 if (encoding == REDIS_ENCODING_INTSET) {
337 ele = createStringObjectFromLongLong(llele);
338 set->ptr = intsetRemove(set->ptr,llele,NULL);
339 } else {
340 incrRefCount(ele);
341 setTypeRemove(set,ele);
342 }
343
344 /* Change argv to replicate as SREM */
345 c->argc = 3;
346 c->argv = zrealloc(c->argv,sizeof(robj*)*(c->argc));
347
348 /* Overwrite SREM with SPOP (same length) */
349 redisAssert(sdslen(c->argv[0]->ptr) == 4);
350 memcpy(c->argv[0]->ptr, "SREM", 4);
351
352 /* Popped element already has incremented refcount */
353 c->argv[2] = ele;
354
355 addReplyBulk(c,ele);
356 if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
357 signalModifiedKey(c->db,c->argv[1]);
358 server.dirty++;
359 }
360
361 void srandmemberCommand(redisClient *c) {
362 robj *set, *ele;
363 int64_t llele;
364 int encoding;
365
366 if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
367 checkType(c,set,REDIS_SET)) return;
368
369 encoding = setTypeRandomElement(set,&ele,&llele);
370 if (encoding == REDIS_ENCODING_INTSET) {
371 addReplyBulkLongLong(c,llele);
372 } else {
373 addReplyBulk(c,ele);
374 }
375 }
376
377 int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
378 return setTypeSize(*(robj**)s1)-setTypeSize(*(robj**)s2);
379 }
380
381 void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
382 robj **sets = zmalloc(sizeof(robj*)*setnum);
383 setTypeIterator *si;
384 robj *eleobj, *dstset = NULL;
385 int64_t intobj;
386 void *replylen = NULL;
387 unsigned long j, cardinality = 0;
388 int encoding;
389
390 for (j = 0; j < setnum; j++) {
391 robj *setobj = dstkey ?
392 lookupKeyWrite(c->db,setkeys[j]) :
393 lookupKeyRead(c->db,setkeys[j]);
394 if (!setobj) {
395 zfree(sets);
396 if (dstkey) {
397 if (dbDelete(c->db,dstkey)) {
398 signalModifiedKey(c->db,dstkey);
399 server.dirty++;
400 }
401 addReply(c,shared.czero);
402 } else {
403 addReply(c,shared.emptymultibulk);
404 }
405 return;
406 }
407 if (checkType(c,setobj,REDIS_SET)) {
408 zfree(sets);
409 return;
410 }
411 sets[j] = setobj;
412 }
413 /* Sort sets from the smallest to largest, this will improve our
414 * algorithm's performace */
415 qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);
416
417 /* The first thing we should output is the total number of elements...
418 * since this is a multi-bulk write, but at this stage we don't know
419 * the intersection set size, so we use a trick, append an empty object
420 * to the output list and save the pointer to later modify it with the
421 * right length */
422 if (!dstkey) {
423 replylen = addDeferredMultiBulkLength(c);
424 } else {
425 /* If we have a target key where to store the resulting set
426 * create this key with an empty set inside */
427 dstset = createIntsetObject();
428 }
429
430 /* Iterate all the elements of the first (smallest) set, and test
431 * the element against all the other sets, if at least one set does
432 * not include the element it is discarded */
433 si = setTypeInitIterator(sets[0]);
434 while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) {
435 for (j = 1; j < setnum; j++) {
436 if (encoding == REDIS_ENCODING_INTSET) {
437 /* intset with intset is simple... and fast */
438 if (sets[j]->encoding == REDIS_ENCODING_INTSET &&
439 !intsetFind((intset*)sets[j]->ptr,intobj))
440 {
441 break;
442 /* in order to compare an integer with an object we
443 * have to use the generic function, creating an object
444 * for this */
445 } else if (sets[j]->encoding == REDIS_ENCODING_HT) {
446 eleobj = createStringObjectFromLongLong(intobj);
447 if (!setTypeIsMember(sets[j],eleobj)) {
448 decrRefCount(eleobj);
449 break;
450 }
451 decrRefCount(eleobj);
452 }
453 } else if (encoding == REDIS_ENCODING_HT) {
454 /* Optimization... if the source object is integer
455 * encoded AND the target set is an intset, we can get
456 * a much faster path. */
457 if (eleobj->encoding == REDIS_ENCODING_INT &&
458 sets[j]->encoding == REDIS_ENCODING_INTSET &&
459 !intsetFind((intset*)sets[j]->ptr,(long)eleobj->ptr))
460 {
461 break;
462 /* else... object to object check is easy as we use the
463 * type agnostic API here. */
464 } else if (!setTypeIsMember(sets[j],eleobj)) {
465 break;
466 }
467 }
468 }
469
470 /* Only take action when all sets contain the member */
471 if (j == setnum) {
472 if (!dstkey) {
473 if (encoding == REDIS_ENCODING_HT)
474 addReplyBulk(c,eleobj);
475 else
476 addReplyBulkLongLong(c,intobj);
477 cardinality++;
478 } else {
479 if (encoding == REDIS_ENCODING_INTSET) {
480 eleobj = createStringObjectFromLongLong(intobj);
481 setTypeAdd(dstset,eleobj);
482 decrRefCount(eleobj);
483 } else {
484 setTypeAdd(dstset,eleobj);
485 }
486 }
487 }
488 }
489 setTypeReleaseIterator(si);
490
491 if (dstkey) {
492 /* Store the resulting set into the target, if the intersection
493 * is not an empty set. */
494 dbDelete(c->db,dstkey);
495 if (setTypeSize(dstset) > 0) {
496 dbAdd(c->db,dstkey,dstset);
497 addReplyLongLong(c,setTypeSize(dstset));
498 } else {
499 decrRefCount(dstset);
500 addReply(c,shared.czero);
501 }
502 signalModifiedKey(c->db,dstkey);
503 server.dirty++;
504 } else {
505 setDeferredMultiBulkLength(c,replylen,cardinality);
506 }
507 zfree(sets);
508 }
509
510 void sinterCommand(redisClient *c) {
511 sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
512 }
513
514 void sinterstoreCommand(redisClient *c) {
515 sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
516 }
517
518 #define REDIS_OP_UNION 0
519 #define REDIS_OP_DIFF 1
520 #define REDIS_OP_INTER 2
521
522 void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op) {
523 robj **sets = zmalloc(sizeof(robj*)*setnum);
524 setTypeIterator *si;
525 robj *ele, *dstset = NULL;
526 int j, cardinality = 0;
527
528 for (j = 0; j < setnum; j++) {
529 robj *setobj = dstkey ?
530 lookupKeyWrite(c->db,setkeys[j]) :
531 lookupKeyRead(c->db,setkeys[j]);
532 if (!setobj) {
533 sets[j] = NULL;
534 continue;
535 }
536 if (checkType(c,setobj,REDIS_SET)) {
537 zfree(sets);
538 return;
539 }
540 sets[j] = setobj;
541 }
542
543 /* We need a temp set object to store our union. If the dstkey
544 * is not NULL (that is, we are inside an SUNIONSTORE operation) then
545 * this set object will be the resulting object to set into the target key*/
546 dstset = createIntsetObject();
547
548 /* Iterate all the elements of all the sets, add every element a single
549 * time to the result set */
550 for (j = 0; j < setnum; j++) {
551 if (op == REDIS_OP_DIFF && j == 0 && !sets[j]) break; /* result set is empty */
552 if (!sets[j]) continue; /* non existing keys are like empty sets */
553
554 si = setTypeInitIterator(sets[j]);
555 while((ele = setTypeNextObject(si)) != NULL) {
556 if (op == REDIS_OP_UNION || j == 0) {
557 if (setTypeAdd(dstset,ele)) {
558 cardinality++;
559 }
560 } else if (op == REDIS_OP_DIFF) {
561 if (setTypeRemove(dstset,ele)) {
562 cardinality--;
563 }
564 }
565 decrRefCount(ele);
566 }
567 setTypeReleaseIterator(si);
568
569 /* Exit when result set is empty. */
570 if (op == REDIS_OP_DIFF && cardinality == 0) break;
571 }
572
573 /* Output the content of the resulting set, if not in STORE mode */
574 if (!dstkey) {
575 addReplyMultiBulkLen(c,cardinality);
576 si = setTypeInitIterator(dstset);
577 while((ele = setTypeNextObject(si)) != NULL) {
578 addReplyBulk(c,ele);
579 decrRefCount(ele);
580 }
581 setTypeReleaseIterator(si);
582 decrRefCount(dstset);
583 } else {
584 /* If we have a target key where to store the resulting set
585 * create this key with the result set inside */
586 dbDelete(c->db,dstkey);
587 if (setTypeSize(dstset) > 0) {
588 dbAdd(c->db,dstkey,dstset);
589 addReplyLongLong(c,setTypeSize(dstset));
590 } else {
591 decrRefCount(dstset);
592 addReply(c,shared.czero);
593 }
594 signalModifiedKey(c->db,dstkey);
595 server.dirty++;
596 }
597 zfree(sets);
598 }
599
600 void sunionCommand(redisClient *c) {
601 sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION);
602 }
603
604 void sunionstoreCommand(redisClient *c) {
605 sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION);
606 }
607
608 void sdiffCommand(redisClient *c) {
609 sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF);
610 }
611
612 void sdiffstoreCommand(redisClient *c) {
613 sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF);
614 }