Use sharding keys to calculate bucket id (WIP)

CRUD allows to automatically calculate `bucket_id` based on primary key
or one can specify `bucket_id` explicitly [1]. However it is often
required to calculate `bucket_id` using sharding keys created by DDL schema.

DDL module exposes space with sharding keys as a part of public API [2],
so everyone is allowed to set and get sharding keys there without adding
DDL module to dependencies.

Patch allows to calculate `bucket_id` value automatically when sharding keys
specified using DDL module or manually in `_ddl_sharding_key` space.

1. #46
2. https://github.com/tarantool/ddl#api
3. #46 (comment)

Closes #166

Author: ligurio

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+* Support calculating `bucket_id` based on `ddl.sharding_key`.
 * Added jsonpath indexes support for queries
 * `tuple-merger` module updated to 0.0.2
 

crud/common/sharding.lua

@@ -7,6 +7,8 @@ local utils = require('crud.common.utils')
 
 local sharding = {}
 
+local sharding_key_cache = nil
+
 function sharding.key_get_bucket_id(key, specified_bucket_id)
     if specified_bucket_id ~= nil then
         return specified_bucket_id
@@ -20,7 +22,17 @@ function sharding.tuple_get_bucket_id(tuple, space, specified_bucket_id)
         return specified_bucket_id
     end
 
-    local key = utils.extract_key(tuple, space.index[0].parts)
+    local primary_index = space.index[0]
+    local key
+    local sharding_key = get_ddl_sharding_key(space.name)
+    if sharding_key ~= nil then
+        key = utils.extract_sharding_key(space, tuple, sharding_key)
+    end
+
+    if key == nil then
+        key = utils.extract_key(tuple, primary_index.parts)
+    end
+
     return sharding.key_get_bucket_id(key)
 end
 
@@ -51,4 +63,22 @@ function sharding.tuple_set_and_return_bucket_id(tuple, space, specified_bucket_
     return bucket_id
 end
 
+-- Get sharding key (actually field names) for all spaces in schema
+-- and cache it's value to speedup access.
+local function sharding.get_ddl_sharding_key(space_name)
+    if box.space._ddl_sharding_key == nil then
+        return nil
+    end
+
+    if sharding_key_cache == nil then
+        sharding_key_cache = box.space._ddl_sharding_key:select{}
+    end
+
+    if space_name ~= nil then
+        return sharding_key_cache[space_name]
+    else
+        return sharding_key_cache
+    end
+end
+
 return sharding

crud/common/utils.lua

@@ -140,6 +140,18 @@ function utils.unflatten(tuple, space_format)
     return object
 end
 
+function utils.extract_sharding_key(space, tuple, sharding_key)
+    local key = {}
+    for i, field in ipairs(space:format()) do
+        for j, name in ipairs(sharding_key) do
+            if name == field.name then
+                key[i] = tuple[i]
+            end
+        end
+    end
+    return key
+end
+
 function utils.extract_key(tuple, key_parts)
     local key = {}
     for i, part in ipairs(key_parts) do
@@ -375,6 +387,24 @@ function utils.get_bucket_id_fieldno(space, shard_index_name)
     return bucket_id_index.parts[1].fieldno
 end
 
+-- NOTE: it's a more general version of get_bucket_id_fieldno()
+-- but it use space's format instead of index
+function utils.get_fieldno_by_name(space_format, field_name)
+    field_name = field_name or 'bucket_id'
+    local field_idx
+    for i, field_format in ipairs(space_format) do
+        if field_format.name == field_name then
+            field_idx = i
+            break
+        end
+    end
+    if field_idx == nil then
+        return nil, ShardingError:new('%q field is not found', field_name)
+    end
+
+    return idx
+end
+
 local uuid_t = ffi.typeof('struct tt_uuid')
 function utils.is_uuid(value)
     return ffi.istype(uuid_t, value)

crud/select/plan.lua

@@ -3,6 +3,7 @@ local errors = require('errors')
 local compare_conditions = require('crud.compare.conditions')
 local utils = require('crud.common.utils')
 local dev_checks = require('crud.common.dev_checks')
+local sharding = require('crud.common.sharding')
 
 local compat = require('crud.common.compat')
 local has_keydef = compat.exists('tuple.keydef', 'key_def')
@@ -16,6 +17,7 @@ local select_plan = {}
 
 local IndexTypeError = errors.new_class('IndexTypeError', {capture_stack = false})
 local FilterFieldsError = errors.new_class('FilterFieldsError',  {capture_stack = false})
+local DDLError = errors.new_class('DDLError', {capture_stack = false})
 
 local function index_is_allowed(index)
     return index.type == 'TREE'
@@ -48,7 +50,7 @@ local function get_index_for_condition(space_indexes, space_format, condition)
     end
 end
 
-local function extract_sharding_key_from_scan_value(scan_value, scan_index, sharding_index)
+local function extract_sharding_key_from_scan_value(scan_value, scan_index, sharding_index, space_format)
     if #scan_value < #sharding_index.parts then
         return nil
     end
@@ -62,6 +64,9 @@ local function extract_sharding_key_from_scan_value(scan_value, scan_index, shar
         scan_value_fields_values[scan_index_part.fieldno] = scan_value[i]
     end
 
+    -- getting sharding_key specified in DDL schema
+    local ddl_sharding_key = sharding.get_ddl_sharding_key(space_name)
+
     -- check that sharding key is included in the scan index fields
     local sharding_key = {}
     for _, sharding_key_part in ipairs(sharding_index.parts) do
@@ -79,7 +84,20 @@ local function extract_sharding_key_from_scan_value(scan_value, scan_index, shar
             return nil
         end
 
-        table.insert(sharding_key, field_value)
+        -- check if a field is a part of DDL sharding key
+        local is_sharding_key_found = false
+        if ddl_sharding_key ~= nil then
+            for i, field_name in ipairs(ddl_sharding_key) do
+                if fieldno == utils.get_fieldno_by_name(space_format, field_name) then
+                    is_sharding_key_found = true
+                    break
+                end
+            end
+        end
+
+        if ddl_sharding_key == nil or is_sharding_key_found == true then
+            table.insert(sharding_key, field_value)
+        end
     end
 
     return sharding_key
@@ -226,12 +244,10 @@ function select_plan.new(space, conditions, opts)
         end
     end
 
-    local sharding_index = primary_index -- XXX: only sharding by primary key is supported
-
-    -- get sharding key value
-    local sharding_key
+    -- getting sharding_key used in primary_index
+    local sharding_index = primary_index
     if scan_value ~= nil and (scan_iter == box.index.EQ or scan_iter == box.index.REQ) then
-        sharding_key = extract_sharding_key_from_scan_value(scan_value, scan_index, sharding_index)
+        sharding_key = extract_sharding_key_from_scan_value(scan_value, scan_index, sharding_index, space_format)
     end
 
     if sharding_key ~= nil and opts.force_map_call ~= true then

test/entrypoint/srv_ddl.lua

@@ -0,0 +1,80 @@
+#!/usr/bin/env tarantool
+
+require('strict').on()
+_G.is_initialized = function() return false end
+
+local log = require('log')
+local errors = require('errors')
+local cartridge = require('cartridge')
+local ddl = require('ddl')
+
+package.preload['customers-storage'] = function()
+    return {
+        role_name = 'customers-storage',
+        init = function()
+            local engine = os.getenv('ENGINE') or 'memtx'
+            local schema_customers = {
+                is_local = true,
+                engine = engine,
+                temporary = false,
+                sharding_key = {
+                    'name',
+                    'age',
+                },
+                format = {
+                    {name = 'id', type = 'unsigned', is_nullable = false},
+                    {name = 'bucket_id', type = 'unsigned', is_nullable = false},
+                    {name = 'name', type = 'string', is_nullable = false},
+                    {name = 'age', type = 'number', is_nullable = false},
+                },
+                indexes = {
+                    {
+                        name = 'primary',
+                        type = 'TREE',
+                        unique = true,
+                        parts = {
+                            {path = 'id', type = 'unsigned', is_nullable = false},
+                        },
+                    },
+                    {
+                        name = 'bucket_id',
+                        type = 'TREE',
+                        unique = false,
+                        parts = {
+                            {path = 'bucket_id', type = 'unsigned', is_nullable = false},
+                        },
+                    },
+                }
+            }
+            local schema = {
+                spaces = {
+                    ['customers'] = schema_customers,
+                },
+            }
+            if not box.cfg.read_only then
+                local ok, err = ddl.set_schema(schema)
+                if not ok then
+                    error(err)
+                end
+            end
+        end,
+    }
+end
+
+local ok, err = errors.pcall('CartridgeCfgError', cartridge.cfg, {
+    advertise_uri = 'localhost:3301',
+    http_port = 8081,
+    bucket_count = 3000,
+    roles = {
+        'customers-storage',
+        'cartridge.roles.crud-router',
+        'cartridge.roles.crud-storage',
+    },
+})
+
+if not ok then
+    log.error('%s', err)
+    os.exit(1)
+end
+
+_G.is_initialized = cartridge.is_healthy