Rewrite client in a register workload

- replace READ operation by a sequence of SQL commands
- replace WRITE operation by a Lua routine with native UPSERT
  and use it as an SQL command
- manage table in client's setup and teardown

Closes #31

Author: ligurio

resources/tarantool/jepsen.lua

@@ -31,9 +31,6 @@ local function bootstrap()
     box.schema.user.grant('jepsen', 'write', 'space', '_schema')
     box.schema.user.grant('jepsen', 'write', 'space', '_space')
 
-    box.schema.space.create('JEPSEN', {engine = '%TARANTOOL_DATA_ENGINE%'})
-    box.space['JEPSEN']:create_index('primary', {parts = {1, 'unsigned' }})
-
 --[[ Function implements a CAS (Compare And Set) operation, which takes a key,
 old value, and new value and sets the key to the new value if and only if the
 old value matches what's currently there, and returns a detailed response
@@ -44,61 +41,39 @@ box.schema.func.create('_CAS',
    {language = 'LUA',
     returns = 'boolean',
     body = [[function(id, old_value, new_value, table)
-    local rc = false
-    box.begin()
-    local tuple = box.space[table]:get{id}
-    if tuple then
-        if tuple[2] == old_value then
-            box.space[table]:update({id}, {{'=', 2, new_value}})
-            rc = true
-        end
-    end
-    box.commit()
+             local rc = false
+             box.begin()
+             local tuple = box.space[table]:get{id}
+             if tuple then
+                 if tuple[2] == old_value then
+                     box.space[table]:update({id}, {{'=', 2, new_value}})
+                     rc = true
+                 end
+             end
+             box.commit()
 
-    return rc
-end]],
+             return rc
+             end]],
     is_sandboxed = false,
     param_list = {'integer', 'integer', 'integer', 'string'},
     exports = {'LUA', 'SQL'},
     is_deterministic = true})
 
---[[ Function implements an WRITE operation, which takes a key and value
-and sets the key to the value if and only if the key is already exists, and
-insert value if it is absent.
-Example: SELECT _WRITE(1, 3, 'JEPSEN')
+--[[ Function implements a UPSERT operation, which takes a key and value
+and sets the key to the value if key exists or insert new key with that value.
+Example: SELECT _UPSERT(1, 3, 4, 'JEPSEN')
 ]]
-box.schema.func.create('_WRITE',
+box.schema.func.create('_UPSERT',
    {language = 'LUA',
-    returns = 'integer',
-    body = [[function (id, value, table)
-             box.space[table]:upsert({id, value}, {{'=', 1, 1}, {'=', 2, value}})
-             return value
+    returns = 'boolean',
+    body = [[function(id, value, table)
+             box.space[table]:upsert({id, value}, {{'=', 2, value}})
+             return true
              end]],
     is_sandboxed = false,
     param_list = {'integer', 'integer', 'string'},
     exports = {'LUA', 'SQL'},
     is_deterministic = true})
-
---[[ Function implements an READ operation, which takes a key and returns a
-value.
-Example: SELECT _READ(1, 'JEPSEN')
-]]
-box.schema.func.create('_READ',
-   {language = 'LUA',
-    returns = 'integer',
-    body = [[function (id, table)
-             box.begin()
-             local tuple = box.space[table]:get{id}
-             if tuple then
-                 return tuple[2]
-             end
-             box.commit()
-             return nil
-             end]],
-    is_sandboxed = false,
-    param_list = {'integer', "string"},
-    exports = {'LUA', 'SQL'},
-    is_deterministic = true})
 end
 
 box.once('jepsen', bootstrap)

src/tarantool/client.clj

@@ -27,22 +27,6 @@
   [node test]
   (j/get-datasource (conn-spec node)))
 
-(defn read-v-by-k
-  "Reads the current value of a key."
-  [conn k]
-  (first (vals (first (j/execute! conn ["SELECT _READ(?, 'JEPSEN')" k])))))
-
-(defn write-v-by-k
-  "Writes the current value of a key."
-  [conn k v]
-  (j/execute! conn ["SELECT _WRITE(?, ?, 'JEPSEN')"
-                    k v]))
-
-(defn compare-and-set
-  [conn id old new]
-  (first (vals (first (j/execute! conn ["SELECT _CAS(?, ?, ?, 'JEPSEN')"
-                                        id old new])))))
-
 (defmacro with-error-handling
   "Common error handling for errors, including txn aborts."
   [op & body]

src/tarantool/register.clj

@@ -12,12 +12,16 @@
                     [nemesis :as nemesis]
                     [tests :as tests]
                     [util :refer [timeout meh]]]
+            [next.jdbc :as j]
+            [next.jdbc.sql :as sql]
             [knossos.model :as model]
             [jepsen.checker.timeline :as timeline]
             [jepsen.os.ubuntu :as ubuntu]
             [tarantool.client :as cl]
             [tarantool.db :as db]))
 
+(def table-name "register")
+
 (defn r   [_ _] {:type :invoke, :f :read, :value nil})
 (defn w   [_ _] {:type :invoke, :f :write, :value (rand-int 5)})
 (defn cas [_ _] {:type :invoke, :f :cas, :value [(rand-int 5) (rand-int 5)]})
@@ -33,27 +37,39 @@
   (setup! [this test node]
     (let [conn (cl/open node test)]
       (assert conn)
-      ;(when (= node (jepsen/primary test))
-      ;  (j/execute! conn ["CREATE TABLE IF NOT EXISTS jepsen (key INT, value INT, PRIMARY KEY (key))"])
-      ;  (j/execute! conn ["CREATE INDEX IF NOT EXISTS idx ON jepsen (key)"]))
+      (cl/with-conn-failure-retry conn
+        (j/execute! conn [(str "CREATE TABLE IF NOT EXISTS " table-name
+                          " (id INT NOT NULL PRIMARY KEY,
+                          value INT NOT NULL)")]))
       (assoc this :conn conn :node node)))
 
   (invoke! [this test op]
      (case (:f op)
-       :read (assoc op
-                    :type  :ok
-                    :value (cl/read-v-by-k conn 1))
-       :write (do (let [con (cl/open (jepsen/primary test) test)]
-                   (cl/write-v-by-k con 1 (:value op)))
-                   (assoc op :type :ok))
+       :read (let [value (:VALUE
+               (first (sql/query conn
+                 [(str "SELECT value FROM " table-name " WHERE id = 1")])))]
+             (assoc op :type :ok :value value))
+
+       :write (do (let [con (cl/open (jepsen/primary test) test)
+                        table (clojure.string/upper-case table-name)]
+                    (j/execute! con
+                      [(str "SELECT _UPSERT(1, " (:value op) ", '" table "')")])
+                  (assoc op :type :ok)))
+
        :cas (let [[old new] (:value op)
-                  con (cl/open (jepsen/primary test) test)]
-                  (assoc op :type (if (cl/compare-and-set con 1 old new)
+                  con (cl/open (jepsen/primary test) test)
+                  table (clojure.string/upper-case table-name)]
+                  (assoc op :type (if (->> (j/execute! conn
+                                             [(str "SELECT _CAS(1, " old ", " new ", '" table "')")])
+                                           (first)
+                                           (vals)
+                                           (first))
                                    :ok
                                    :fail)))))
 
-  (teardown! [this test])
-      ;(j/execute! conn ["DROP TABLE jepsen"]))
+  (teardown! [_ test]
+    (cl/with-conn-failure-retry conn
+      (j/execute! conn [(str "DROP TABLE IF EXISTS " table-name)])))
 
   (close! [_ test]))