Allow to specify bucket_id

* Bucket ID can be specified using `opts.bucket_id` for all operations
* For `insert`/`insert_object` bucket ID can be specified as
  tuple/object field as well as `opts.bucket_id` value

Author: dokshina

README.md

@@ -15,8 +15,11 @@ across the cluster.
 **Notes:**
 
 * A space should have a format.
-* `bucket_id` is computed as `vshard.router.bucket_id_strcrc32(key)`,
+* By default, `bucket_id` is computed as `vshard.router.bucket_id_strcrc32(key)`,
   where `key` is the primary key value.
+  Custom bucket ID can be specified as `opts.bucket_id` for each operation.
+  For `insert`/`insert_object` bucket ID can be specified as
+  tuple/object field as well as `opts.bucket_id` value.
 
 ### Insert
 
@@ -33,6 +36,7 @@ where:
 * `tuple` / `object` (`table`) - tuple/object to insert
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns metadata and array contains one inserted row, error.
 
@@ -75,6 +79,7 @@ where:
 * `key` (`any`) - primary key value
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns metadata and array contains one row, error.
 
@@ -106,6 +111,7 @@ where:
 * `operations` (`table`) - update [operations](https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_space/#box-space-update)
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns metadata and array contains one updated row, error.
 
@@ -136,6 +142,7 @@ where:
 * `key` (`any`) - primary key value
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns metadata and array contains one deleted row (empty for vinyl), error.
 
@@ -168,6 +175,7 @@ where:
 * `tuple` / `object` (`table`) - tuple/object to insert or replace exist one
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns inserted or replaced rows and metadata or nil with error.
 
@@ -214,6 +222,7 @@ where:
 * `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)
+  * `bucket_id` (`?number|cdata`) - bucket ID
 
 Returns metadata and empty array of rows or nil, error.
 
@@ -267,6 +276,8 @@ where:
   * `after` (`?table`) - tuple after which objects should be selected
   * `batch_size` (`?number`) - number of tuples to process per one request to storage
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `bucket_id` (`?number|cdata`) - bucket ID
+    (is used when select by full primary key is performed)
 
 Returns metadata and array of rows, error.
 

crud/common/sharding.lua

@@ -0,0 +1,24 @@
+local vshard = require('vshard')
+
+local utils = require('crud.common.utils')
+
+local sharding = {}
+
+function sharding.get_bucket_id_by_key(key, specified_bucket_id)
+    if specified_bucket_id ~= nil then
+        return specified_bucket_id
+    end
+
+    return vshard.router.bucket_id_strcrc32(key)
+end
+
+function sharding.get_bucket_id_by_tuple(tuple, space, specified_bucket_id)
+    if specified_bucket_id ~= nil then
+        return specified_bucket_id
+    end
+
+    local key = utils.extract_key(tuple, space.index[0].parts)
+    return sharding.get_bucket_id_by_key(key)
+end
+
+return sharding

crud/delete.lua

@@ -5,6 +5,7 @@ local vshard = require('vshard')
 local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
+local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local DeleteError = errors.new_class('Delete',  {capture_stack = false})
@@ -44,18 +45,22 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.bucket_id
+--  Bucket ID
+--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
+--
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function delete.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
 
-
     local space = utils.get_space(space_name, vshard.router.routeall())
     if space == nil then
         return nil, DeleteError:new("Space %q doesn't exist", space_name)
@@ -65,7 +70,7 @@ function delete.call(space_name, key, opts)
         key = key:totable()
     end
 
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local bucket_id = sharding.get_bucket_id_by_key(key, opts.bucket_id)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
         return nil, DeleteError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)

crud/get.lua

@@ -5,6 +5,7 @@ local vshard = require('vshard')
 local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
+local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local GetError = errors.new_class('Get',  {capture_stack = false})
@@ -44,13 +45,18 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.bucket_id
+--  Bucket ID
+--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
+--
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function get.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -64,8 +70,7 @@ function get.call(space_name, key, opts)
         key = key:totable()
     end
 
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
-
+    local bucket_id = sharding.get_bucket_id_by_key(key, opts.bucket_id)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
         return nil, GetError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)

crud/insert.lua

@@ -5,6 +5,7 @@ local vshard = require('vshard')
 local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
+local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local InsertError = errors.new_class('Insert',  {capture_stack = false})
@@ -43,13 +44,18 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.bucket_id
+--  Bucket ID
+--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
+--
 -- @return[1] tuple
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function insert.tuple(space_name, tuple, opts)
     checks('string', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -60,21 +66,32 @@ function insert.tuple(space_name, tuple, opts)
     end
     local space_format = space:format()
 
-    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, InsertError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
-    end
-
     local bucket_id_fieldno, err = utils.get_bucket_id_fieldno(space)
     if err ~= nil then
         return nil, err
     end
 
-    if tuple[bucket_id_fieldno] ~= nil then
-        return nil, InsertError:new("Unexpected value (%s) at field %s (bucket_id)",
-                tuple[bucket_id_fieldno], bucket_id_fieldno)
+    if opts.bucket_id ~= nil then
+        if tuple[bucket_id_fieldno] == nil then
+            tuple[bucket_id_fieldno] = opts.bucket_id
+        else
+            if tuple[bucket_id_fieldno] ~= opts.bucket_id then
+                return nil, InsertError:new(
+                    "Tuple and opts.bucket_id contain different bucket_id values: %s and %s",
+                    tuple[bucket_id_fieldno], opts.bucket_id
+                )
+            end
+        end
+    end
+
+    if tuple[bucket_id_fieldno] == nil then
+        tuple[bucket_id_fieldno] = sharding.get_bucket_id_by_tuple(tuple, space)
+    end
+
+    local bucket_id = tuple[bucket_id_fieldno]
+    local replicaset, err = vshard.router.route(bucket_id)
+    if replicaset == nil then
+        return nil, InsertError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
     end
 
     tuple[bucket_id_fieldno] = bucket_id
@@ -104,17 +121,15 @@ end
 -- @param table obj
 --  Object
 --
--- @tparam ?number opts.timeout
---  Function call timeout
+-- @tparam ?table opts
+--  Options of insert.tuple
 --
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function insert.object(space_name, obj, opts)
-    checks('string', 'table', {
-        timeout = '?number',
-    })
+    checks('string', 'table', '?table')
 
     opts = opts or {}
 

crud/replace.lua

@@ -5,6 +5,7 @@ local vshard = require('vshard')
 local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
+local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local ReplaceError = errors.new_class('Replace', { capture_stack = false })
@@ -43,13 +44,18 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.bucket_id
+--  Bucket ID
+--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
+--
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function replace.tuple(space_name, tuple, opts)
     checks('string', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -59,9 +65,7 @@ function replace.tuple(space_name, tuple, opts)
         return nil, ReplaceError:new("Space %q doesn't exist", space_name)
     end
 
-    local key = utils.extract_key(tuple, space.index[0].parts)
-
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local bucket_id = sharding.get_bucket_id_by_tuple(tuple, space, opts.bucket_id)
     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)
@@ -104,17 +108,15 @@ end
 -- @param table obj
 --  Object
 --
--- @tparam ?number opts.timeout
---  Function call timeout
+-- @tparam ?table opts
+--  Options of replace.tuple
 --
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function replace.object(space_name, obj, opts)
-    checks('string', 'table', {
-        timeout = '?number',
-    })
+    checks('string', 'table', '?table')
 
     opts = opts or {}
 

crud/select.lua

@@ -5,6 +5,7 @@ local vshard = require('vshard')
 local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
+local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local select_conditions = require('crud.select.conditions')
@@ -104,8 +105,7 @@ local function select_iteration(space_name, plan, opts)
     return results
 end
 
-local function get_replicasets_by_sharding_key(sharding_key)
-    local bucket_id = vshard.router.bucket_id_strcrc32(sharding_key)
+local function get_replicasets_by_sharding_key(bucket_id)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
         return nil, GetReplicasetsError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
@@ -122,6 +122,7 @@ local function build_select_iterator(space_name, user_conditions, opts)
         first = '?number',
         timeout = '?number',
         batch_size = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -163,7 +164,8 @@ local function build_select_iterator(space_name, user_conditions, opts)
     local replicasets_to_select = replicasets
 
     if plan.sharding_key ~= nil then
-        replicasets_to_select = get_replicasets_by_sharding_key(plan.sharding_key)
+        local bucket_id = sharding.get_bucket_id_by_key(plan.sharding_key, opts.bucket_id)
+        replicasets_to_select = get_replicasets_by_sharding_key(bucket_id)
     end
 
     -- generate tuples comparator
@@ -202,6 +204,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         timeout = '?number',
         batch_size = '?number',
         use_tomap = '?boolean',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -215,6 +218,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         first = opts.first,
         timeout = opts.timeout,
         batch_size = opts.batch_size,
+        bucket_id = opts.bucket_id,
     })
 
     if err ~= nil then
@@ -251,6 +255,7 @@ function select_module.call(space_name, user_conditions, opts)
         first = '?number',
         timeout = '?number',
         batch_size = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -266,6 +271,7 @@ function select_module.call(space_name, user_conditions, opts)
         first = opts.first,
         timeout = opts.timeout,
         batch_size = opts.batch_size,
+        bucket_id = opts.bucket_id,
     })
 
     if err ~= nil then