Fix filling sharding key cache when sharding key data is incorrect

Sharding key cache contains sharding key structures
separated by space name. Before this fix if sharding key
was incorrect for some space an error was returned and
sharding keys that were after the incorrect one
did not get into the cache. The solution is to
create a variable for an error and write a
message to it if an error occurred for the space
we are working with, output a warning for other spaces.

Part of #237

Author: AnaNek

crud/common/sharding/sharding_key.lua

@@ -1,4 +1,5 @@
 local errors = require('errors')
+local log = require('log')
 
 local dev_checks = require('crud.common.dev_checks')
 local cache = require('crud.common.sharding.sharding_metadata_cache')
@@ -96,8 +97,10 @@ function sharding_key_module.extract_from_pk(space_name, sharding_key_as_index_o
     return extract_from_index(primary_key, primary_index_parts, sharding_key_as_index_obj)
 end
 
-function sharding_key_module.construct_as_index_obj_cache(metadata_map)
-    dev_checks('table')
+function sharding_key_module.construct_as_index_obj_cache(metadata_map, specified_space_name)
+    dev_checks('table', 'string')
+
+    local result_err
 
     cache.sharding_key_as_index_obj_map = {}
     for space_name, metadata in pairs(metadata_map) do
@@ -106,12 +109,19 @@ function sharding_key_module.construct_as_index_obj_cache(metadata_map)
                                                                    metadata.space_format,
                                                                    metadata.sharding_key_def)
             if err ~= nil then
-                return err
+                if specified_space_name == space_name then
+                    result_err = err
+                    log.error(err)
+                else
+                    log.warn(err)
+                end
             end
 
             cache.sharding_key_as_index_obj_map[space_name] = sharding_key_as_index_obj
         end
     end
+
+    return result_err
 end
 
 sharding_key_module.internal = {

crud/common/sharding/sharding_metadata.lua

@@ -68,8 +68,8 @@ end
 -- a sharding metadata by a single one, other fibers will wait while
 -- cache.fetch_lock become unlocked during timeout passed to
 -- _fetch_on_router().
-local _fetch_on_router = locked(function(timeout, metadata_map_name)
-    dev_checks('number', 'string')
+local _fetch_on_router = locked(function(timeout, space_name, metadata_map_name)
+    dev_checks('number', 'string', 'string')
 
     if cache[metadata_map_name] ~= nil then
         return
@@ -86,7 +86,7 @@ local _fetch_on_router = locked(function(timeout, metadata_map_name)
         return
     end
 
-    local err = sharding_key.construct_as_index_obj_cache(metadata_map)
+    local err = sharding_key.construct_as_index_obj_cache(metadata_map, space_name)
     if err ~= nil then
         return err
     end
@@ -107,11 +107,8 @@ local function fetch_on_router(space_name, metadata_map_name, timeout)
     end
 
     local timeout = timeout or const.FETCH_SHARDING_METADATA_TIMEOUT
-    local err = _fetch_on_router(timeout, metadata_map_name)
+    local err = _fetch_on_router(timeout, space_name, metadata_map_name)
     if err ~= nil then
-        if cache[metadata_map_name] ~= nil then
-            return cache[metadata_map_name][space_name]
-        end
         return nil, err
     end
 

test/helper.lua

@@ -332,4 +332,12 @@ function helpers.update_sharding_key_cache(cluster, space_name)
     ]], {space_name})
 end
 
+function helpers.get_sharding_key_cache(cluster)
+    return cluster.main_server.net_box:eval([[
+        local sharding_metadata_cache = require('crud.common.sharding.sharding_metadata_cache')
+
+        return sharding_metadata_cache[sharding_metadata_cache.SHARDING_KEY_MAP_NAME]
+    ]])
+end
+
 return helpers

test/integration/ddl_sharding_key_test.lua

@@ -694,3 +694,70 @@ pgroup.test_update_cache = function(g)
     t.assert_equals(err, nil)
     t.assert_equals(sharding_key_as_index_obj, {parts = {{fieldno = 3}}})
 end
+
+pgroup.test_update_cache_with_incorrect_key = function(g)
+    -- get data from cache for space with correct sharding key
+    local space_name = 'customers_name_key'
+
+    local sharding_key_as_index_obj, err = helpers.update_sharding_key_cache(g.cluster, space_name)
+    t.assert_equals(err, nil)
+    t.assert_equals(sharding_key_as_index_obj, {parts = {{fieldno = 3}}})
+
+    -- records for all spaces exist
+    sharding_key_as_index_obj = helpers.get_sharding_key_cache(g.cluster)
+    t.assert_equals(sharding_key_as_index_obj, {
+        customers_age_key = {parts = {{fieldno = 4}}},
+        customers_name_age_key_different_indexes = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_age_key_three_fields_index = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_key = {parts = {{fieldno = 3}}},
+        customers_name_key_non_uniq_index = {parts = {{fieldno = 3}}},
+        customers_name_key_uniq_index = {parts = {{fieldno = 3}}},
+        customers_secondary_idx_name_key = {parts = {{fieldno = 3}}},
+    })
+
+    -- no error just warning
+    local space_name = 'customers_name_key'
+    helpers.call_on_servers(g.cluster, {'s1-master', 's2-master'}, function(server)
+        server.net_box:call('set_sharding_key', {space_name, {'non_existent_field'}})
+    end)
+
+    -- we get no error because we sent request for correct space
+    local sharding_key_as_index_obj, err = helpers.update_sharding_key_cache(g.cluster, 'customers_age_key')
+    t.assert_equals(err, nil)
+    t.assert_equals(sharding_key_as_index_obj, {parts = {{fieldno = 4}}})
+
+    -- cache['customers_name_key'] == nil (space with incorrect key)
+    -- other records for correct spaces exist in cache
+    sharding_key_as_index_obj = helpers.get_sharding_key_cache(g.cluster)
+    t.assert_equals(sharding_key_as_index_obj, {
+        customers_age_key = {parts = {{fieldno = 4}}},
+        customers_name_age_key_different_indexes = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_age_key_three_fields_index = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_key_non_uniq_index = {parts = {{fieldno = 3}}},
+        customers_name_key_uniq_index = {parts = {{fieldno = 3}}},
+        customers_secondary_idx_name_key = {parts = {{fieldno = 3}}},
+    })
+
+    -- get data from cache for space with incorrect sharding key
+    local space_name = 'customers_name_key'
+    helpers.call_on_servers(g.cluster, {'s1-master', 's2-master'}, function(server)
+        server.net_box:call('set_sharding_key', {space_name, {'non_existent_field'}})
+    end)
+
+    -- we get an error because we sent request for incorrect space
+    local sharding_key_as_index_obj, err = helpers.update_sharding_key_cache(g.cluster, space_name)
+    t.assert_equals(sharding_key_as_index_obj, nil)
+    t.assert_str_contains(err.err, "No such field (non_existent_field) in a space format (customers_name_key)")
+
+    -- cache['customers_name_key'] == nil (space with incorrect key)
+    -- other records for correct spaces exist in cache
+    sharding_key_as_index_obj = helpers.get_sharding_key_cache(g.cluster)
+    t.assert_equals(sharding_key_as_index_obj, {
+        customers_age_key = {parts = {{fieldno = 4}}},
+        customers_name_age_key_different_indexes = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_age_key_three_fields_index = {parts = {{fieldno = 3}, {fieldno = 4}}},
+        customers_name_key_non_uniq_index = {parts = {{fieldno = 3}}},
+        customers_name_key_uniq_index = {parts = {{fieldno = 3}}},
+        customers_secondary_idx_name_key = {parts = {{fieldno = 3}}},
+    })
+end