Fixed RESTORE hash failure (Issue #532)

[redis.git] / src / t_hash.c

diff --git a/src/t_hash.c b/src/t_hash.c
index 9699223a8f02dd36e44fae6bbb6095e42b199631..5b7a347abbcd75379ac1e314862f8c83ff3c4c97 100644 (file)
--- a/src/t_hash.c
+++ b/src/t_hash.c
@@ -47,23 +47,18 @@ int hashTypeGetFromZiplist(robj *o, robj *field,
 
     zl = o->ptr;
     fptr = ziplistIndex(zl, ZIPLIST_HEAD);
 
     zl = o->ptr;
     fptr = ziplistIndex(zl, ZIPLIST_HEAD);

-    while (fptr != NULL) {
-        /* Grab pointer to the value (fptr points to the field) */
-        vptr = ziplistNext(zl, fptr);
-        redisAssert(vptr != NULL);
-
-        /* Compare field in ziplist with specified field */
-        if (ziplistCompare(fptr, field->ptr, sdslen(field->ptr))) {
-            break;
+    if (fptr != NULL) {
+        fptr = ziplistFind(fptr, field->ptr, sdslen(field->ptr), 1);
+        if (fptr != NULL) {
+            /* Grab pointer to the value (fptr points to the field) */
+            vptr = ziplistNext(zl, fptr);
+            redisAssert(vptr != NULL);

         }
         }

-
-        /* Skip over value */
-        fptr = ziplistNext(zl, vptr);

     }
 
     decrRefCount(field);
 
     }
 
     decrRefCount(field);
 

-    if (fptr != NULL) {
+    if (vptr != NULL) {

         ret = ziplistGet(vptr, vstr, vlen, vll);
         redisAssert(ret);
         return 0;
         ret = ziplistGet(vptr, vstr, vlen, vll);
         redisAssert(ret);
         return 0;


@@ -80,10 +75,7 @@ int hashTypeGetFromHashTable(robj *o, robj *field, robj **value) {
     redisAssert(o->encoding == REDIS_ENCODING_HT);
 
     de = dictFind(o->ptr, field);
     redisAssert(o->encoding == REDIS_ENCODING_HT);
 
     de = dictFind(o->ptr, field);

-    if (de == NULL) {
-        return -1;
-    }
-
+    if (de == NULL) return -1;

     *value = dictGetVal(de);
     return 0;
 }
     *value = dictGetVal(de);
     return 0;
 }


@@ -117,11 +109,9 @@ robj *hashTypeGetObject(robj *o, robj *field) {
             incrRefCount(aux);
             value = aux;
         }
             incrRefCount(aux);
             value = aux;
         }

-

     } else {
         redisPanic("Unknown hash encoding");
     }
     } else {
         redisPanic("Unknown hash encoding");
     }

-

     return value;
 }
 
     return value;
 }
 


@@ -133,26 +123,21 @@ int hashTypeExists(robj *o, robj *field) {
         unsigned int vlen = UINT_MAX;
         long long vll = LLONG_MAX;
 
         unsigned int vlen = UINT_MAX;
         long long vll = LLONG_MAX;
 

-        if (hashTypeGetFromZiplist(o, field, &vstr, &vlen, &vll) == 0) {
-            return 1;
-        }
-
+        if (hashTypeGetFromZiplist(o, field, &vstr, &vlen, &vll) == 0) return 1;

     } else if (o->encoding == REDIS_ENCODING_HT) {
         robj *aux;
 
     } else if (o->encoding == REDIS_ENCODING_HT) {
         robj *aux;
 

-        if (hashTypeGetFromHashTable(o, field, &aux) == 0) {
-            return 1;
-        }
-
+        if (hashTypeGetFromHashTable(o, field, &aux) == 0) return 1;

     } else {
         redisPanic("Unknown hash encoding");
     }
     } else {
         redisPanic("Unknown hash encoding");
     }

-

     return 0;
 }
 
 /* Add an element, discard the old if the key already exists.
     return 0;
 }
 
 /* Add an element, discard the old if the key already exists.

- * Return 0 on insert and 1 on update. */
+ * Return 0 on insert and 1 on update.
+ * This function will take care of incrementing the reference count of the
+ * retained fields and value objects. */

 int hashTypeSet(robj *o, robj *field, robj *value) {
     int update = 0;
 
 int hashTypeSet(robj *o, robj *field, robj *value) {
     int update = 0;
 


@@ -164,50 +149,44 @@ int hashTypeSet(robj *o, robj *field, robj *value) {
 
         zl = o->ptr;
         fptr = ziplistIndex(zl, ZIPLIST_HEAD);
 
         zl = o->ptr;
         fptr = ziplistIndex(zl, ZIPLIST_HEAD);

-        while (fptr != NULL) {
-            /* Compare field in ziplist with specified field */
-            if (ziplistCompare(fptr, field->ptr, sdslen(field->ptr))) {
-                zl = ziplistDelete(zl,&fptr);
-                zl = ziplistDelete(zl,&fptr);
-                o->ptr = zl;
+        if (fptr != NULL) {
+            fptr = ziplistFind(fptr, field->ptr, sdslen(field->ptr), 1);
+            if (fptr != NULL) {
+                /* Grab pointer to the value (fptr points to the field) */
+                vptr = ziplistNext(zl, fptr);
+                redisAssert(vptr != NULL);

                 update = 1;
                 update = 1;

-                break;
-            }

 
 

-            /* Grab pointer to the value (fptr points to the field) */
-            vptr = ziplistNext(zl, fptr);
-            redisAssert(vptr != NULL);
+                /* Delete value */
+                zl = ziplistDelete(zl, &vptr);

 
 

-            /* Grab pointer (if any) to the next field */
-            fptr = ziplistNext(zl, vptr);
+                /* Insert new value */
+                zl = ziplistInsert(zl, vptr, value->ptr, sdslen(value->ptr));
+            }

         }
 
         }
 

-        /* Push new field/value pair onto the tail of the ziplist */
-        zl = ziplistPush(zl, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
-        zl = ziplistPush(zl, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL);
+        if (!update) {
+            /* Push new field/value pair onto the tail of the ziplist */
+            zl = ziplistPush(zl, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
+            zl = ziplistPush(zl, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL);
+        }

         o->ptr = zl;
         o->ptr = zl;

-

         decrRefCount(field);
         decrRefCount(value);
 
         /* Check if the ziplist needs to be converted to a hash table */
         decrRefCount(field);
         decrRefCount(value);
 
         /* Check if the ziplist needs to be converted to a hash table */

-        if (hashTypeLength(o) > server.hash_max_ziplist_entries) {
+        if (hashTypeLength(o) > server.hash_max_ziplist_entries)

             hashTypeConvert(o, REDIS_ENCODING_HT);
             hashTypeConvert(o, REDIS_ENCODING_HT);

-        }
-

     } else if (o->encoding == REDIS_ENCODING_HT) {
         if (dictReplace(o->ptr, field, value)) { /* Insert */
             incrRefCount(field);
         } else { /* Update */
             update = 1;
         }
     } else if (o->encoding == REDIS_ENCODING_HT) {
         if (dictReplace(o->ptr, field, value)) { /* Insert */
             incrRefCount(field);
         } else { /* Update */
             update = 1;
         }

-

         incrRefCount(value);
         incrRefCount(value);

-

     } else {
         redisPanic("Unknown hash encoding");
     }
     } else {
         redisPanic("Unknown hash encoding");
     }

-

     return update;
 }
 
     return update;
 }
 


@@ -217,28 +196,20 @@ int hashTypeDelete(robj *o, robj *field) {
     int deleted = 0;
 
     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
     int deleted = 0;
 
     if (o->encoding == REDIS_ENCODING_ZIPLIST) {

-        unsigned char *zl, *fptr, *vptr;
+        unsigned char *zl, *fptr;

 
         field = getDecodedObject(field);
 
         zl = o->ptr;
         fptr = ziplistIndex(zl, ZIPLIST_HEAD);
 
         field = getDecodedObject(field);
 
         zl = o->ptr;
         fptr = ziplistIndex(zl, ZIPLIST_HEAD);

-        while (fptr != NULL) {
-            /* Compare field in ziplist with specified field */
-            if (ziplistCompare(fptr, field->ptr, sdslen(field->ptr))) {
+        if (fptr != NULL) {
+            fptr = ziplistFind(fptr, field->ptr, sdslen(field->ptr), 1);
+            if (fptr != NULL) {

                 zl = ziplistDelete(zl,&fptr);
                 zl = ziplistDelete(zl,&fptr);
                 o->ptr = zl;
                 deleted = 1;
                 zl = ziplistDelete(zl,&fptr);
                 zl = ziplistDelete(zl,&fptr);
                 o->ptr = zl;
                 deleted = 1;

-                break;

             }
             }

-
-            /* Grab pointer to the value (fptr points to the field) */
-            vptr = ziplistNext(zl, fptr);
-            redisAssert(vptr != NULL);
-
-            /* Grab pointer (if any) to the next field */
-            fptr = ziplistNext(zl, vptr);

         }
 
         decrRefCount(field);
         }
 
         decrRefCount(field);


@@ -318,10 +289,7 @@ int hashTypeNext(hashTypeIterator *hi) {
             redisAssert(vptr != NULL);
             fptr = ziplistNext(zl, vptr);
         }
             redisAssert(vptr != NULL);
             fptr = ziplistNext(zl, vptr);
         }

-
-        if (fptr == NULL) {
-            return REDIS_ERR;
-        }
+        if (fptr == NULL) return REDIS_ERR;

 
         /* Grab pointer to the value (fptr points to the field) */
         vptr = ziplistNext(zl, fptr);
 
         /* Grab pointer to the value (fptr points to the field) */
         vptr = ziplistNext(zl, fptr);


@@ -330,16 +298,11 @@ int hashTypeNext(hashTypeIterator *hi) {
         /* fptr, vptr now point to the first or next pair */
         hi->fptr = fptr;
         hi->vptr = vptr;
         /* fptr, vptr now point to the first or next pair */
         hi->fptr = fptr;
         hi->vptr = vptr;

-

     } else if (hi->encoding == REDIS_ENCODING_HT) {
     } else if (hi->encoding == REDIS_ENCODING_HT) {

-        if ((hi->de = dictNext(hi->di)) == NULL) {
-            return REDIS_ERR;
-        }
-
+        if ((hi->de = dictNext(hi->di)) == NULL) return REDIS_ERR;

     } else {
         redisPanic("Unknown hash encoding");
     }
     } else {
         redisPanic("Unknown hash encoding");
     }

-

     return REDIS_OK;
 }
 
     return REDIS_OK;
 }
 


@@ -518,7 +481,7 @@ void hmsetCommand(redisClient *c) {
 }
 
 void hincrbyCommand(redisClient *c) {
 }
 
 void hincrbyCommand(redisClient *c) {

-    long long value, incr;
+    long long value, incr, oldvalue;

     robj *o, *current, *new;
 
     if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;
     robj *o, *current, *new;
 
     if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;


@@ -534,6 +497,12 @@ void hincrbyCommand(redisClient *c) {
         value = 0;
     }
 
         value = 0;
     }
 

+    oldvalue = value;
+    if ((incr < 0 && oldvalue < 0 && incr < (LLONG_MIN-oldvalue)) ||
+        (incr > 0 && oldvalue > 0 && incr > (LLONG_MAX-oldvalue))) {
+        addReplyError(c,"increment or decrement would overflow");
+        return;
+    }

     value += incr;
     new = createStringObjectFromLongLong(value);
     hashTypeTryObjectEncoding(o,&c->argv[2],NULL);
     value += incr;
     new = createStringObjectFromLongLong(value);
     hashTypeTryObjectEncoding(o,&c->argv[2],NULL);


@@ -546,7 +515,7 @@ void hincrbyCommand(redisClient *c) {
 
 void hincrbyfloatCommand(redisClient *c) {
     double long value, incr;
 
 void hincrbyfloatCommand(redisClient *c) {
     double long value, incr;

-    robj *o, *current, *new;
+    robj *o, *current, *new, *aux;

 
     if (getLongDoubleFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;
     if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
 
     if (getLongDoubleFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;
     if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;


@@ -566,9 +535,17 @@ void hincrbyfloatCommand(redisClient *c) {
     hashTypeTryObjectEncoding(o,&c->argv[2],NULL);
     hashTypeSet(o,c->argv[2],new);
     addReplyBulk(c,new);
     hashTypeTryObjectEncoding(o,&c->argv[2],NULL);
     hashTypeSet(o,c->argv[2],new);
     addReplyBulk(c,new);

-    decrRefCount(new);

     signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;
     signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;

+
+    /* Always replicate HINCRBYFLOAT as an HSET command with the final value
+     * in order to make sure that differences in float pricision or formatting
+     * will not create differences in replicas or after an AOF restart. */
+    aux = createStringObject("HSET",4);
+    rewriteClientCommandArgument(c,0,aux);
+    decrRefCount(aux);
+    rewriteClientCommandArgument(c,3,new);
+    decrRefCount(new);

 }
 
 static void addHashFieldToReply(redisClient *c, robj *o, robj *field) {
 }
 
 static void addHashFieldToReply(redisClient *c, robj *o, robj *field) {