Support replace and upsert operations (#11)

Co-authored-by: savolgin <[email protected]>

Author: savolgin

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+* CRUD operations:
+ * replace
+ * upsert
+
 ## [0.1.0] - 2020-09-23
 
 ### Added

README.md

@@ -129,6 +129,61 @@ crud.delete('customers', 1)
 ...
 ```
 
+### Replace
+
+```lua
+local object, err = crud.replace(space_name, object, opts)
+```
+
+where:
+
+* `space_name` (`string`) - name of the space
+* `object` (`table`) - object to insert or replace exist one
+* `opts`:
+  * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+
+Returns inserted or replaced object, error.
+
+**Example:**
+
+```lua
+crud.replace('customers', {
+    id = 1, name = 'Alice', age = 22,
+})
+---
+- bucket_id: 7614
+  age: 22
+  name: Alice
+  id: 1
+...
+```
+
+### Upsert
+
+```lua
+local object, err = crud.upsert(space_name, object, operations, opts)
+```
+
+where:
+
+* `space_name` (`string`) - name of the space
+* `object` (`table`) - object to insert if there is no existing tuple which matches the key fields
+* `operations` (`table`) - update [operations](https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/#box-space-update) if there is an existing tuple which matches the key fields of tuple
+* `opts`:
+  * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+
+Returns nil, error.
+
+**Example:**
+
+```lua
+crud.upsert('customers', {id = 1, name = 'Alice', age = 22,}, {{'+', 'age', 1}})
+---
+- nil
+...
+```
+
+
 ### Select
 
 `CRUD` supports multi-conditional selects, treating a cluster as a single space.

crud.lua

@@ -5,8 +5,10 @@
 local registry = require('crud.common.registry')
 local call = require('crud.common.call')
 local insert = require('crud.insert')
+local replace = require('crud.replace')
 local get = require('crud.get')
 local update = require('crud.update')
+local upsert = require('crud.upsert')
 local delete = require('crud.delete')
 local select = require('crud.select')
 
@@ -30,10 +32,18 @@ crud.insert = insert.call
 -- @function get
 crud.get = get.call
 
+-- @refer replace.call
+-- @function replace
+crud.replace = replace.call
+
 -- @refer update.call
 -- @function update
 crud.update = update.call
 
+-- @refer upsert.call
+-- @function upsert
+crud.upsert = upsert.call
+
 -- @refer delete.call
 -- @function delete
 crud.delete = delete.call
@@ -57,7 +67,9 @@ function crud.init()
     call.init()
     insert.init()
     get.init()
+    replace.init()
     update.init()
+    upsert.init()
     delete.init()
     select.init()
 end

crud/common/utils.lua

@@ -3,6 +3,7 @@ local errors = require('errors')
 
 local FlattenError = errors.new_class("FlattenError", {capture_stack = false})
 local UnflattenError = errors.new_class("UnflattenError", {capture_stack = false})
+local ParseOperationsError = errors.new_class('ParseOperationsError',  {capture_stack = false})
 
 local utils = {}
 
@@ -35,7 +36,7 @@ end
 
 local system_fields = { bucket_id = true }
 
-function utils.flatten(object, space_format)
+function utils.flatten(object, space_format, bucket_id)
     if object == nil then return nil end
 
     local tuple = {}
@@ -49,6 +50,10 @@ function utils.flatten(object, space_format)
             end
         end
 
+        if bucket_id ~= nil and field_format.name == 'bucket_id' then
+            value = bucket_id
+        end
+
         tuple[fieldno] = value
     end
 
@@ -99,4 +104,56 @@ function utils.merge_primary_key_parts(key_parts, pk_parts)
     return merged_parts
 end
 
+local __tarantool_supports_fieldpaths
+local function tarantool_supports_fieldpaths()
+    if __tarantool_supports_fieldpaths ~= nil then
+        return __tarantool_supports_fieldpaths
+    end
+
+    local major_minor_patch = _G._TARANTOOL:split('-', 1)[1]
+    local major_minor_patch_parts = major_minor_patch:split('.', 2)
+
+    local major = tonumber(major_minor_patch_parts[1])
+    local minor = tonumber(major_minor_patch_parts[2])
+    local patch = tonumber(major_minor_patch_parts[3])
+
+    -- since Tarantool 2.3
+    __tarantool_supports_fieldpaths = major >= 2 and (minor > 3 or minor == 3 and patch >= 1)
+
+    return __tarantool_supports_fieldpaths
+end
+
+function utils.convert_operations(user_operations, space_format)
+    if tarantool_supports_fieldpaths() then
+        return user_operations
+    end
+
+    local converted_operations = {}
+
+    for _, operation in ipairs(user_operations) do
+        if type(operation[2]) == 'string' then
+            local field_id
+            for fieldno, field_format in ipairs(space_format) do
+                if field_format.name == operation[2] then
+                    field_id = fieldno
+                    break
+                end
+            end
+
+            if field_id == nil then
+                return nil, ParseOperationsError:new(
+                        "Space format doesn't contain field named %q", operation[2])
+            end
+
+            table.insert(converted_operations, {
+                operation[1], field_id, operation[3]
+            })
+        else
+            table.insert(converted_operations, operation)
+        end
+    end
+
+    return converted_operations
+end
+
 return utils

crud/insert.lua

@@ -62,9 +62,10 @@ function insert.call(space_name, obj, opts)
     if space == nil then
         return nil, InsertError:new("Space %q doesn't exists", space_name)
     end
+    local space_format = space:format()
 
     -- compute default buckect_id
-    local tuple, err = utils.flatten(obj, space:format(), true)
+    local tuple, err = utils.flatten(obj, space_format)
     if err ~= nil then
         return nil, InsertError:new("Object is specified in bad format: %s", err)
     end
@@ -77,10 +78,7 @@ function insert.call(space_name, obj, opts)
         return nil, InsertError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
     end
 
-    obj = table.copy(obj)
-    obj.bucket_id = bucket_id
-
-    local tuple, err = utils.flatten(obj, space:format())
+    local tuple, err = utils.flatten(obj, space_format, bucket_id)
     if err ~= nil then
         return nil, InsertError:new("Object is specified in bad format: %s", err)
     end
@@ -95,7 +93,7 @@ function insert.call(space_name, obj, opts)
     end
 
     local tuple = results[replicaset.uuid]
-    local object, err = utils.unflatten(tuple, space:format())
+    local object, err = utils.unflatten(tuple, space_format)
     if err ~= nil then
         return nil, InsertError:new("Received tuple that doesn't match space format: %s", err)
     end

crud/replace.lua

@@ -0,0 +1,101 @@
+local checks = require('checks')
+local errors = require('errors')
+local vshard = require('vshard')
+
+local call = require('crud.common.call')
+local registry = require('crud.common.registry')
+local utils = require('crud.common.utils')
+
+require('crud.common.checkers')
+
+local ReplaceError = errors.new_class('Replace', { capture_stack = false })
+
+local replace = {}
+
+local REPLACE_FUNC_NAME = '__replace'
+
+local function call_replace_on_storage(space_name, tuple)
+    checks('string', 'table')
+
+    local space = box.space[space_name]
+    if space == nil then
+        return nil, ReplaceError:new("Space %q doesn't exists", space_name)
+    end
+
+    return space:replace(tuple)
+end
+
+function replace.init()
+    registry.add({
+        [REPLACE_FUNC_NAME] = call_replace_on_storage,
+    })
+end
+
+--- Insert or replace a tuple in the specifed space
+--
+-- @function call
+--
+-- @param string space_name
+--  A space name
+--
+-- @param table obj
+--  Tuple object (according to space format)
+--
+-- @tparam ?number opts.timeout
+--  Function call timeout
+--
+-- @return[1] object
+-- @treturn[2] nil
+-- @treturn[2] table Error description
+--
+function replace.call(space_name, obj, opts)
+    checks('string', 'table', {
+        timeout = '?number',
+    })
+
+    opts = opts or {}
+
+    local space = utils.get_space(space_name, vshard.router.routeall())
+    if space == nil then
+        return nil, ReplaceError:new("Space %q doesn't exists", space_name)
+    end
+
+    local space_format = space:format()
+    -- compute default buckect_id
+    local tuple, err = utils.flatten(obj, space_format)
+    if err ~= nil then
+        return nil, ReplaceError:new("Object is specified in bad format: %s", err)
+    end
+
+    local key = utils.extract_key(tuple, space.index[0].parts)
+
+    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local replicaset, err = vshard.router.route(bucket_id)
+    if replicaset == nil then
+        return nil, ReplaceError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
+    end
+
+    local tuple, err = utils.flatten(obj, space_format, bucket_id)
+    if err ~= nil then
+        return nil, ReplaceError:new("Object is specified in bad format: %s", err)
+    end
+
+    local results, err = call.rw(REPLACE_FUNC_NAME, {space_name, tuple}, {
+        replicasets = {replicaset},
+        timeout = opts.timeout,
+    })
+
+    if err ~= nil then
+        return nil, ReplaceError:new("Failed to replace: %s", err)
+    end
+
+    local tuple = results[replicaset.uuid]
+    local object, err = utils.unflatten(tuple, space_format)
+    if err ~= nil then
+        return nil, ReplaceError:new("Received tuple that doesn't match space format: %s", err)
+    end
+
+    return object
+end
+
+return replace