Support sharding by custom bucket_id

Author: dokshina

README.md

@@ -15,8 +15,9 @@ 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.
 
 ### Insert
 
@@ -33,6 +34,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 +77,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 +109,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 +140,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 +173,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 +220,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 +274,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,16 @@
+local vshard = require('vshard')
+
+local utils = require('crud.common.utils')
+
+local sharding = {}
+
+function sharding.get_bucket_id_by_key(key, _)
+    return vshard.router.bucket_id_strcrc32(key)
+end
+
+function sharding.get_bucket_id_by_tuple(tuple, space)
+    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})
@@ -51,11 +52,11 @@ end
 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 +66,7 @@ function delete.call(space_name, key, opts)
         key = key:totable()
     end
 
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local bucket_id = opts.bucket_id or sharding.get_bucket_id_by_key(key, space)
     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})
@@ -51,6 +52,7 @@ end
 function get.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -64,8 +66,7 @@ function get.call(space_name, key, opts)
         key = key:totable()
     end
 
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
-
+    local bucket_id = opts.bucket_id or sharding.get_bucket_id_by_key(key, space)
     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})
@@ -50,6 +51,7 @@ end
 function insert.tuple(space_name, tuple, opts)
     checks('string', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -60,8 +62,7 @@ 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 bucket_id = opts.bucket_id or sharding.get_bucket_id_by_tuple(tuple, space)
     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)
@@ -104,17 +105,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 })
@@ -50,6 +51,7 @@ end
 function replace.tuple(space_name, tuple, opts)
     checks('string', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -59,9 +61,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 = opts.bucket_id or sharding.get_bucket_id_by_tuple(tuple, space)
     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 +104,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 = opts.bucket_id or sharding.get_bucket_id_by_key(plan.sharding_key, space)
+        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

crud/update.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 UpdateError = errors.new_class('Update',  {capture_stack = false})
@@ -55,6 +56,7 @@ end
 function update.call(space_name, key, user_operations, opts)
     checks('string', '?', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -74,7 +76,7 @@ function update.call(space_name, key, user_operations, opts)
         return nil, UpdateError:new("Wrong operations are specified: %s", err)
     end
 
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local bucket_id = opts.bucket_id or sharding.get_bucket_id_by_key(key, space)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
         return nil, UpdateError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)

crud/upsert.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 UpsertError = errors.new_class('UpsertError',  { capture_stack = false})
@@ -54,6 +55,7 @@ end
 function upsert.tuple(space_name, tuple, user_operations, opts)
     checks('string', '?', 'table', {
         timeout = '?number',
+        bucket_id = '?number|cdata',
     })
 
     opts = opts or {}
@@ -69,9 +71,7 @@ function upsert.tuple(space_name, tuple, user_operations, opts)
         return nil, UpsertError:new("Wrong operations are specified: %s", err)
     end
 
-    local key = utils.extract_key(tuple, space.index[0].parts)
-
-    local bucket_id = vshard.router.bucket_id_strcrc32(key)
+    local bucket_id = opts.bucket_id or sharding.get_bucket_id_by_tuple(tuple, space)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
         return nil, UpsertError:new("Failed to get replicaset for bucket_id %s: %s", bucket_id, err.err)
@@ -118,17 +118,15 @@ end
 --  user_operations to be performed.
 --  See `space:upsert()` operations in Tarantool doc
 --
--- @tparam ?number opts.timeout
---  Function call timeout
+-- @tparam ?table opts
+--  Options of upsert.tuple
 --
 -- @return[1] object
 -- @treturn[2] nil
 -- @treturn[2] table Error description
 --
 function upsert.object(space_name, obj, user_operations, opts)
-    checks('string', '?', 'table', {
-        timeout = '?number',
-    })
+    checks('string', 'table', 'table', '?table')
 
     opts = opts or {}