Use DDL sharding key in select operation

Part of #166

Author: ligurio

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
@@ -51,4 +53,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.
+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

@@ -375,6 +375,24 @@ function utils.get_bucket_id_fieldno(space, shard_index_name)
     return bucket_id_index.parts[1].fieldno
 end
 
+-- NOTE: it is like get_bucket_id_fieldno()
+-- but it uses 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 in a space format', field_name)
+    end
+
+    return field_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')
@@ -48,7 +49,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, space_name)
     if #scan_value < #sharding_index.parts then
         return nil
     end
@@ -62,6 +63,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 +83,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 _, 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 +243,13 @@ function select_plan.new(space, conditions, opts)
         end
     end
 
-    local sharding_index = primary_index -- XXX: only sharding by primary key is supported
-
+    -- getting sharding_key used in primary_index
+    local sharding_index = primary_index
     -- get sharding key value
     local sharding_key
     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, space_name)
     end
 
     if sharding_key ~= nil and opts.force_map_call ~= true then