WIP: compare hashes

Author: DifferentialOrange

crud/common/sharding/init.lua

@@ -3,10 +3,12 @@ local errors = require('errors')
 
 local BucketIDError = errors.new_class("BucketIDError", {capture_stack = false})
 local GetReplicasetsError = errors.new_class('GetReplicasetsError', {capture_stack = false})
+local ShardingHashMismatchError = errors.new_class("ShardingHashMismatchError", {capture_stack = false})
 
 local utils = require('crud.common.utils')
 local dev_checks = require('crud.common.dev_checks')
 local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
+local storage_metadata_cache = require('crud.common.sharding.storage_metadata_cache')
 
 local sharding = {}
 
@@ -25,37 +27,49 @@ function sharding.key_get_bucket_id(space_name, key, specified_bucket_id)
     dev_checks('string', '?', '?number|cdata')
 
     if specified_bucket_id ~= nil then
-        return specified_bucket_id
+        return { bucket_id = specified_bucket_id }
     end
 
-    local sharding_func, err = sharding_metadata_module.fetch_sharding_func_on_router(space_name)
+    local sharding_func_data, err = sharding_metadata_module.fetch_sharding_func_on_router(space_name)
     if err ~= nil then
         return nil, err
     end
 
-    if sharding_func ~= nil then
-        return sharding_func(key)
+    if sharding_func_data.value ~= nil then
+        return {
+            bucket_id = sharding_func_data.value(key),
+            sharding_func_hash = sharding_func_data.hash,
+        }
     end
 
-    return vshard.router.bucket_id_strcrc32(key)
+    return { bucket_id = vshard.router.bucket_id_strcrc32(key) }
 end
 
 function sharding.tuple_get_bucket_id(tuple, space, specified_bucket_id)
     if specified_bucket_id ~= nil then
-        return specified_bucket_id
+        return { bucket_id = specified_bucket_id }
     end
 
     local sharding_index_parts = space.index[0].parts
-    local sharding_key_as_index_obj, err = sharding_metadata_module.fetch_sharding_key_on_router(space.name)
+    local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space.name)
     if err ~= nil then
         return nil, err
     end
-    if sharding_key_as_index_obj ~= nil then
-        sharding_index_parts = sharding_key_as_index_obj.parts
+    if sharding_key_data.value ~= nil then
+        sharding_index_parts = sharding_key_data.value.parts
     end
     local key = utils.extract_key(tuple, sharding_index_parts)
 
-    return sharding.key_get_bucket_id(space.name, key)
+    local bucket_id_data, err = sharding.key_get_bucket_id(space.name, key, nil)
+    if err ~= nil then
+        return nil, err
+    end
+
+    return {
+        bucket_id = bucket_id_data.bucket_id,
+        sharding_func_hash = bucket_id_data.sharding_func_hash,
+        sharding_key_hash = sharding_key_data.hash
+    }
 end
 
 function sharding.tuple_set_and_return_bucket_id(tuple, space, specified_bucket_id)
@@ -77,16 +91,35 @@ function sharding.tuple_set_and_return_bucket_id(tuple, space, specified_bucket_
         end
     end
 
-    local bucket_id = tuple[bucket_id_fieldno]
-    if bucket_id == nil then
-        bucket_id, err = sharding.tuple_get_bucket_id(tuple, space)
+    local sharding_data = { bucket_id = tuple[bucket_id_fieldno] }
+
+    if sharding_data.bucket_id == nil then
+        sharding_data, err = sharding.tuple_get_bucket_id(tuple, space)
         if err ~= nil then
             return nil, err
         end
-        tuple[bucket_id_fieldno] = bucket_id
+        tuple[bucket_id_fieldno] = sharding_data.bucket_id
+    end
+
+    return sharding_data
+end
+
+function sharding.check_sharding_hash(space_name, sharding_func_hash, sharding_key_hash, skip_sharding_hash_check)
+    if skip_sharding_hash_check == true then
+        return true
+    end
+
+    local storage_func_hash = storage_metadata_cache.get_sharding_func_hash(space_name)
+    local storage_key_hash = storage_metadata_cache.get_sharding_key_hash(space_name)
+
+    if storage_func_hash ~= sharding_func_hash or storage_key_hash ~= sharding_key_hash then
+        local err_msg = ('crud: Sharding hash mismatch for space %s. ' ..
+                         'Sharding data will be reloaded after receiving this error. ' ..
+                         'Please retry your request.'):format(space_name)
+        return nil, ShardingHashMismatchError:new(err_msg)
     end
 
-    return bucket_id
+    return true
 end
 
 return sharding

crud/common/sharding/sharding_metadata.lua

@@ -121,7 +121,10 @@ end)
 
 local function fetch_on_router(space_name, metadata_map_name, timeout)
     if cache[metadata_map_name] ~= nil then
-        return cache[metadata_map_name][space_name]
+        return {
+            value = cache[metadata_map_name][space_name],
+            hash = cache[cache.META_HASH_MAP_NAME][metadata_map_name][space_name]
+        }
     end
 
     local timeout = timeout or const.FETCH_SHARDING_METADATA_TIMEOUT
@@ -131,7 +134,10 @@ local function fetch_on_router(space_name, metadata_map_name, timeout)
     end
 
     if cache[metadata_map_name] ~= nil then
-        return cache[metadata_map_name][space_name]
+        return {
+            value = cache[metadata_map_name][space_name],
+            hash = cache[cache.META_HASH_MAP_NAME][metadata_map_name][space_name],
+        }
     end
 
     return nil, FetchShardingMetadataError:new(

crud/count.lua

@@ -27,6 +27,9 @@ local function count_on_storage(space_name, index_id, conditions, opts)
         tarantool_iter = 'number',
         yield_every = '?number',
         scan_condition_num = '?number',
+        sharding_func_hash = '?number',
+        sharding_key_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
     })
 
     opts = opts or {}
@@ -38,6 +41,14 @@ local function count_on_storage(space_name, index_id, conditions, opts)
         return nil, CountError:new("Index with ID %s doesn't exist", index_id)
     end
 
+    local _, err = sharding.check_sharding_hash(space_name,
+                                                opts.sharding_func_hash,
+                                                opts.sharding_key_hash,
+                                                opts.skip_sharding_hash_check)
+    if err ~= nil then
+        return nil, err
+    end
+
     local value = opts.scan_value
 
     local filter_func, err = filters.gen_func(space, conditions, {
@@ -114,14 +125,14 @@ local function call_count_on_router(space_name, user_conditions, opts)
         return nil, CountError:new("Space %q doesn't exist", space_name), true
     end
 
-    local sharding_key_as_index_obj, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
+    local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
     if err ~= nil then
         return nil, err
     end
 
     -- plan count
     local plan, err = count_plan.new(space, conditions, {
-        sharding_key_as_index_obj = sharding_key_as_index_obj,
+        sharding_key_as_index_obj = sharding_key_data.value,
     })
     if err ~= nil then
         return nil, CountError:new("Failed to plan count: %s", err), true
@@ -159,21 +170,28 @@ local function call_count_on_router(space_name, user_conditions, opts)
     --       eye to resharding. However, AFAIU, the optimization
     --       does not make the result less consistent (sounds
     --       weird, huh?).
+    local sharding_func_hash = nil
+    local skip_sharding_hash_check = nil
+
     local perform_map_reduce = opts.force_map_call == true or
         (opts.bucket_id == nil and plan.sharding_key == nil)
     if not perform_map_reduce then
-        local bucket_id, err = sharding.key_get_bucket_id(space_name, plan.sharding_key, opts.bucket_id)
+        local bucket_id_data, err = sharding.key_get_bucket_id(space_name, plan.sharding_key, opts.bucket_id)
         if err ~= nil then
             return nil, err
         end
 
-        assert(bucket_id ~= nil)
+        assert(bucket_id_data.bucket_id ~= nil)
+
+        sharding_func_hash = bucket_id_data.sharding_func_hash
 
         local err
-        replicasets_to_count, err = sharding.get_replicasets_by_bucket_id(bucket_id)
+        replicasets_to_count, err = sharding.get_replicasets_by_bucket_id(bucket_id_data.bucket_id)
         if err ~= nil then
             return nil, err, true
         end
+    else
+        skip_sharding_hash_check = true
     end
 
     local yield_every = opts.yield_every or DEFAULT_YIELD_EVERY
@@ -191,6 +209,9 @@ local function call_count_on_router(space_name, user_conditions, opts)
         tarantool_iter = plan.tarantool_iter,
         yield_every = yield_every,
         scan_condition_num = plan.scan_condition_num,
+        sharding_func_hash = sharding_func_hash,
+        sharding_key_hash = sharding_key_data.hash,
+        skip_sharding_hash_check = skip_sharding_hash_check,
     }
 
     local results, err = call.map(COUNT_FUNC_NAME, {

crud/delete.lua

@@ -16,14 +16,29 @@ local delete = {}
 
 local DELETE_FUNC_NAME = '_crud.delete_on_storage'
 
-local function delete_on_storage(space_name, key, field_names)
-    dev_checks('string', '?', '?table')
+local function delete_on_storage(space_name, key, field_names, opts)
+    dev_checks('string', '?', '?table', {
+        sharding_key_hash = '?number',
+        sharding_func_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
+    })
+
+    opts = opts or {}
 
     local space = box.space[space_name]
     if space == nil then
         return nil, DeleteError:new("Space %q doesn't exist", space_name)
     end
 
+    local _, err = sharding.check_sharding_hash(space_name,
+                                                opts.sharding_func_hash,
+                                                opts.sharding_key_hash,
+                                                opts.skip_sharding_hash_check)
+
+    if err ~= nil then
+        return nil, err
+    end
+
     -- add_space_schema_hash is false because
     -- reloading space format on router can't avoid delete error on storage
     return schema.wrap_box_space_func_result(space, 'delete', {key}, {
@@ -57,35 +72,49 @@ local function call_delete_on_router(space_name, key, opts)
         key = key:totable()
     end
 
+    local sharding_key_hash = nil
+    local skip_sharding_hash_check = nil
+
     local sharding_key = key
     if opts.bucket_id == nil then
         local primary_index_parts = space.index[0].parts
 
-        local sharding_key_as_index_obj, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
+        local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
         if err ~= nil then
             return nil, err
         end
 
         sharding_key, err = sharding_key_module.extract_from_pk(space_name,
-                                                                sharding_key_as_index_obj,
+                                                                sharding_key_data.value,
                                                                 primary_index_parts, key)
         if err ~= nil then
             return nil, err
         end
+
+        sharding_key_hash = sharding_key_data.hash
+    else
+        skip_sharding_hash_check = true
     end
 
-    local bucket_id, err = sharding.key_get_bucket_id(space_name, sharding_key, opts.bucket_id)
+    local bucket_id_data, err = sharding.key_get_bucket_id(space_name, sharding_key, opts.bucket_id)
     if err ~= nil then
         return nil, err
     end
 
+    local delete_on_storage_opts = {
+        sharding_func_hash = bucket_id_data.sharding_func_hash,
+        sharding_key_hash = sharding_key_hash,
+        skip_sharding_hash_check = skip_sharding_hash_check,
+    }
+
     local call_opts = {
         mode = 'write',
         timeout = opts.timeout,
     }
+
     local storage_result, err = call.single(
-        bucket_id, DELETE_FUNC_NAME,
-        {space_name, key, opts.fields},
+        bucket_id_data.bucket_id, DELETE_FUNC_NAME,
+        {space_name, key, opts.fields, delete_on_storage_opts},
         call_opts
     )
 

crud/get.lua

@@ -16,14 +16,29 @@ local get = {}
 
 local GET_FUNC_NAME = '_crud.get_on_storage'
 
-local function get_on_storage(space_name, key, field_names)
-    dev_checks('string', '?', '?table')
+local function get_on_storage(space_name, key, field_names, opts)
+    dev_checks('string', '?', '?table', {
+        sharding_key_hash = '?number',
+        sharding_func_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
+    })
+
+    opts = opts or {}
 
     local space = box.space[space_name]
     if space == nil then
         return nil, GetError:new("Space %q doesn't exist", space_name)
     end
 
+    local _, err = sharding.check_sharding_hash(space_name,
+                                                opts.sharding_func_hash,
+                                                opts.sharding_key_hash,
+                                                opts.skip_sharding_hash_check)
+
+    if err ~= nil then
+        return nil, err
+    end
+
     -- add_space_schema_hash is false because
     -- reloading space format on router can't avoid get error on storage
     return schema.wrap_box_space_func_result(space, 'get', {key}, {
@@ -61,36 +76,50 @@ local function call_get_on_router(space_name, key, opts)
     end
 
     local sharding_key = key
+    local sharding_key_hash = nil
+    local skip_sharding_hash_check = nil
+
     if opts.bucket_id == nil then
         local primary_index_parts = space.index[0].parts
 
-        local sharding_key_as_index_obj, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
+        local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
         if err ~= nil then
             return nil, err
         end
 
         sharding_key, err = sharding_key_module.extract_from_pk(space_name,
-                                                                sharding_key_as_index_obj,
+                                                                sharding_key_data.value,
                                                                 primary_index_parts, key)
         if err ~= nil then
             return nil, err
         end
+
+        sharding_key_hash = sharding_key_data.hash
+    else
+        skip_sharding_hash_check = true
     end
 
-    local bucket_id, err = sharding.key_get_bucket_id(space_name, sharding_key, opts.bucket_id)
+    local bucket_id_data, err = sharding.key_get_bucket_id(space_name, sharding_key, opts.bucket_id)
     if err ~= nil then
         return nil, err
     end
 
+    local get_on_storage_opts = {
+        sharding_func_hash = bucket_id_data.sharding_func_hash,
+        sharding_key_hash = sharding_key_hash,
+        skip_sharding_hash_check = skip_sharding_hash_check,
+    }
+
     local call_opts = {
         mode = opts.mode or 'read',
         prefer_replica = opts.prefer_replica,
         balance = opts.balance,
         timeout = opts.timeout,
     }
+
     local storage_result, err = call.single(
-        bucket_id, GET_FUNC_NAME,
-        {space_name, key, opts.fields},
+        bucket_id_data.bucket_id, GET_FUNC_NAME,
+        {space_name, key, opts.fields, get_on_storage_opts},
         call_opts
     )