RONDB-768: Modified handling of SET commands

pink/rondis/common.cc

@@ -25,3 +25,19 @@ void assign_generic_err_to_response(
     std::cout << buf;
     response->assign(buf);
 }
+
+void set_length(char *buf, Uint32 key_len)
+{
+    Uint8 *ptr = (Uint8 *)buf;
+    ptr[0] = (Uint8)(key_len & 255);
+    ptr[1] = (Uint8)(key_len >> 8);
+}
+
+Uint32 get_length(char *buf)
+{
+    Uint8 *ptr = (Uint8 *)buf;
+    Uint8 low = ptr[0];
+    Uint8 high = ptr[1];
+    Uint32 len32 = Uint32(low) + Uint32(256) * Uint32(high);
+    return len32;
+}

pink/rondis/common.h

@@ -16,6 +16,8 @@
 int write_formatted(char *buffer, int bufferSize, const char *format, ...);
 void assign_ndb_err_to_response(std::string *response, const char *app_str, NdbError error);
 void assign_generic_err_to_response(std::string *response, const char *app_str);
+void set_length(char* buf, Uint32 key_len);
+Uint32 get_length(char* buf);
 
 // NDB API error messages
 #define FAILED_GET_DICT "Failed to get NdbDict"

pink/rondis/string/commands.cc

@@ -162,6 +162,7 @@ void rondb_set(
     Uint32 value_len = argv[arg_index_start + 1].size();
     char varsize_param[EXTENSION_VALUE_LEN + 500];
     Uint32 num_value_rows = 0;
+    Uint32 prev_num_rows = 0;
     Uint64 rondb_key = 0;
 
     if (value_len > INLINE_VALUE_LEN)
@@ -191,7 +192,6 @@ void rondb_set(
 
     int ret_code = 0;
     ret_code = create_key_row(response,
-                              ndb,
                               tab,
                               trans,
                               redis_key_id,
@@ -201,8 +201,8 @@ void rondb_set(
                               value_str,
                               value_len,
                               num_value_rows,
-                              Uint32(0),
-                              &varsize_param[0]);
+                              prev_num_rows,
+                              Uint32(0));
     if (ret_code != 0)
     {
         // Often unnecessary since it already failed to commit
@@ -226,41 +226,68 @@ void rondb_set(
             assign_ndb_err_to_response(response, FAILED_CREATE_TXN_OBJECT, ndb->getNdbError());
             return;
         }
-        if (delete_and_insert_key_row(response,
-                                      ndb,
-                                      tab,
-                                      trans,
-                                      redis_key_id,
-                                      rondb_key,
-                                      key_str,
-                                      key_len,
-                                      value_str,
-                                      value_len,
-                                      num_value_rows,
-                                      Uint32(0),
-                                      &varsize_param[0]) != 0)
-        {
-            ndb->closeTransaction(trans);
-            return;
-        }
-    }
-
-    if (num_value_rows == 0)
-    {
+        /**
+         * We don't know the exact number of value rows, but we know that it is
+         * at least one.
+         */
+        prev_num_rows = 1;
+        ret_code = create_key_row(response,
+                                  tab,
+                                  trans,
+                                  redis_key_id,
+                                  rondb_key,
+                                  key_str,
+                                  key_len,
+                                  value_str,
+                                  value_len,
+                                  num_value_rows,
+                                  prev_num_rows,
+                                  Uint32(0));
+    } else if (num_value_rows == 0) {
         ndb->closeTransaction(trans);
         response->append("+OK\r\n");
         return;
     }
-    create_all_value_rows(response,
-                          ndb,
-                          dict,
-                          trans,
-                          rondb_key,
-                          value_str,
-                          value_len,
-                          num_value_rows,
-                          &varsize_param[0]);
+    /**
+     * Coming here means that we either have to add new value rows or we have
+     * to delete previous value rows or both. Thus the transaction is still
+     * open. We start by creating the new value rows. Next we delete the
+     * remaining value rows from the previous instantiation of the row.
+     */
+    if (num_value_rows > 0) {
+        ret_code = create_all_value_rows(response,
+                                         ndb,
+                                         dict,
+                                         trans,
+                                         rondb_key,
+                                         value_str,
+                                         value_len,
+                                         num_value_rows,
+                                         &varsize_param[0]);
+    }
+    if (ret_code != 0) {
+        return;
+    }
+    ret_code = delete_value_rows(response,
+                                 tab,
+                                 trans,
+                                 rondb_key,
+                                 num_value_rows,
+                                 prev_num_rows);
+    if (ret_code != 0) {
+        return;
+    }
+    if (trans->execute(NdbTransaction::Commit,
+                       NdbOperation::AbortOnError) == 0 &&
+        trans->getNdbError().code != 0)
+    {
+        assign_ndb_err_to_response(response,
+                                   FAILED_EXEC_TXN,
+                                   trans->getNdbError());
+        return;
+    }
     ndb->closeTransaction(trans);
+    response->append("+OK\r\n");
     return;
 }
 

pink/rondis/string/commands.h

@@ -22,9 +22,6 @@
     Most importantly, it writes Ndb error messages to the response string. This may
     however change in the future, since this causes redundancy.
 */
-void set_length(char* buf, Uint32 key_len);
-Uint32 get_length(char* buf);
-
 void rondb_get_command(Ndb *ndb,
                        const pink::RedisCmdArgsType &argv,
                        std::string *response);