ddl: fix error on explicit bucket_id

After introducing sharding hash info comparison [1], requests with ddl
and explicit bucket_id in options started to fail with
"Sharding hash mismatch" error. It affected following methods:
- insert
- insert_object
- replace
- replace_object
- upsert
- upsert_object
- count

The situation is as follows. Due to a code mistake, router hasn't passed
a sharding hash with a request if bucket_id was specified. If there was
any ddl information for a space on storage, it has caused a hash
mismatch error. Since sharding info reload couldn't fix broken hash
extraction, request failed after a number of retries. This patch fixes
this behavior by skipping hash comparison if sharding info wasn't used
(we already do it in other methods).

1. #268

Closes #278

Author: DifferentialOrange

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Changed
 
 ### Fixed
+* Requests no more fail with "Sharding hash mismatch" error
+  if ddl set and bucket_id is explicitly specified (#278).
 
 ## [0.11.0] - 20-04-22
 

README.md

@@ -83,6 +83,10 @@ with sharding func definition as a part of
 or insert manually to the space `_ddl_sharding_func`.
 
 Automatic sharding key and function reload is supported since version 0.11.0.
+Version 0.11.0 contains critical bug that causes some CRUD methods to fail
+with "Sharding hash mismatch" error if ddl is set and bucket_id is provided
+explicitly ([#278](https://github.com/tarantool/crud/issues/278)). Please,
+upgrade to 0.11.1 instead.
 
 CRUD uses `strcrc32` as sharding function by default.
 The reason why using of `strcrc32` is undesirable is that

crud/common/sharding/init.lua

@@ -100,6 +100,8 @@ function sharding.tuple_set_and_return_bucket_id(tuple, space, specified_bucket_
             return nil, err
         end
         tuple[bucket_id_fieldno] = sharding_data.bucket_id
+    else
+        sharding_data.skip_sharding_hash_check = true
     end
 
     return sharding_data

crud/count.lua

@@ -126,9 +126,18 @@ local function call_count_on_router(space_name, user_conditions, opts)
         return nil, CountError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
     end
 
-    local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
-    if err ~= nil then
-        return nil, err
+    local sharding_key_data = {}
+    local sharding_func_hash = nil
+    local skip_sharding_hash_check = nil
+
+    -- We don't need sharding info if bucket_id specified.
+    if opts.bucket_id == nil then
+        sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(space_name)
+        if err ~= nil then
+            return nil, err
+        end
+    else
+        skip_sharding_hash_check = true
     end
 
     -- plan count
@@ -171,8 +180,6 @@ 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)

crud/insert.lua

@@ -21,6 +21,7 @@ local function insert_on_storage(space_name, tuple, opts)
         fields = '?table',
         sharding_key_hash = '?number',
         sharding_func_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
     })
 
     opts = opts or {}
@@ -82,6 +83,7 @@ local function call_insert_on_router(space_name, original_tuple, opts)
         fields = opts.fields,
         sharding_func_hash = sharding_data.sharding_func_hash,
         sharding_key_hash = sharding_data.sharding_key_hash,
+        skip_sharding_hash_check = sharding_data.skip_sharding_hash_check,
     }
 
     local call_opts = {

crud/replace.lua

@@ -21,6 +21,7 @@ local function replace_on_storage(space_name, tuple, opts)
         fields = '?table',
         sharding_key_hash = '?number',
         sharding_func_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
     })
 
     opts = opts or {}
@@ -86,6 +87,7 @@ local function call_replace_on_router(space_name, original_tuple, opts)
         fields = opts.fields,
         sharding_func_hash = sharding_data.sharding_func_hash,
         sharding_key_hash = sharding_data.sharding_key_hash,
+        skip_sharding_hash_check = sharding_data.skip_sharding_hash_check,
     }
 
     local call_opts = {

crud/upsert.lua

@@ -20,6 +20,7 @@ local function upsert_on_storage(space_name, tuple, operations, opts)
         add_space_schema_hash = '?boolean',
         sharding_key_hash = '?number',
         sharding_func_hash = '?number',
+        skip_sharding_hash_check = '?boolean',
     })
 
     opts = opts or {}
@@ -93,6 +94,7 @@ local function call_upsert_on_router(space_name, original_tuple, user_operations
         fields = opts.fields,
         sharding_func_hash = sharding_data.sharding_func_hash,
         sharding_key_hash = sharding_data.sharding_key_hash,
+        skip_sharding_hash_check = sharding_data.skip_sharding_hash_check,
     }
 
     local call_opts = {

test/integration/ddl_sharding_func_test.lua

@@ -540,3 +540,203 @@ cache_group.test_update_cache_with_incorrect_func = function(g)
     cache_size = helpers.get_sharding_func_cache_size(g.cluster)
     t.assert_equals(cache_size, 1)
 end
+
+
+local known_bucket_id_key = {1, 'Emma'}
+local known_bucket_id_tuple = {
+    known_bucket_id_key[1],
+    box.NULL,
+    known_bucket_id_key[2],
+    22
+}
+local known_bucket_id_object = {
+    id = known_bucket_id_key[1],
+    bucket_id = box.NULL,
+    name = known_bucket_id_key[2],
+    age = 22
+}
+local known_bucket_id = 1111
+local known_bucket_id_result_tuple = {
+    known_bucket_id_key[1],
+    known_bucket_id,
+    known_bucket_id_key[2],
+    22
+}
+local known_bucket_id_result = {
+    s1 = nil,
+    s2 = known_bucket_id_result_tuple,
+}
+local known_bucket_id_update = {{'+', 'age', 1}}
+local known_bucket_id_updated_result = {
+    s1 = nil,
+    s2 = {known_bucket_id_key[1], known_bucket_id, known_bucket_id_key[2], 23},
+}
+local prepare_known_bucket_id_data = function(g)
+    if known_bucket_id_result.s1 ~= nil then
+        local conn_s1 = g.cluster:server('s1-master').net_box
+        local result = conn_s1.space[g.params.space_name]:insert(known_bucket_id_result.s1)
+        t.assert_equals(result, known_bucket_id_result.s1)
+    end
+
+    if known_bucket_id_result.s2 ~= nil then
+        local conn_s2 = g.cluster:server('s2-master').net_box
+        local result = conn_s2.space[g.params.space_name]:insert(known_bucket_id_result.s2)
+        t.assert_equals(result, known_bucket_id_result.s2)
+    end
+end
+
+local known_bucket_id_write_cases = {
+    insert = {
+        func = 'crud.insert',
+        input_2 = known_bucket_id_tuple,
+        input_3 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    insert_object = {
+        func = 'crud.insert_object',
+        input_2 = known_bucket_id_object,
+        input_3 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    replace = {
+        func = 'crud.replace',
+        input_2 = known_bucket_id_tuple,
+        input_3 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    replace_object = {
+        func = 'crud.replace_object',
+        input_2 = known_bucket_id_object,
+        input_3 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    upsert = {
+        func = 'crud.upsert',
+        input_2 = known_bucket_id_tuple,
+        input_3 = {},
+        input_4 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    upsert_object = {
+        func = 'crud.upsert_object',
+        input_2 = known_bucket_id_object,
+        input_3 = {},
+        input_4 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_result,
+    },
+    update = {
+        before_test = prepare_known_bucket_id_data,
+        func = 'crud.update',
+        input_2 = known_bucket_id_key,
+        input_3 = known_bucket_id_update,
+        input_4 = {bucket_id = known_bucket_id},
+        result = known_bucket_id_updated_result,
+    },
+    delete = {
+        before_test = prepare_known_bucket_id_data,
+        func = 'crud.delete',
+        input_2 = known_bucket_id_key,
+        input_3 = {bucket_id = known_bucket_id},
+        result = {},
+    },
+}
+
+for name, case in pairs(known_bucket_id_write_cases) do
+    local test_name = ('test_gh_278_%s_with_explicit_bucket_id_and_ddl'):format(name)
+
+    if case.before_test ~= nil then
+        pgroup.before_test(test_name, case.before_test)
+    end
+
+    pgroup[test_name] = function(g)
+        local obj, err = g.cluster.main_server.net_box:call(
+            case.func, {
+                g.params.space_name,
+                case.input_2,
+                case.input_3,
+                case.input_4,
+            })
+        t.assert_equals(err, nil)
+        t.assert_is_not(obj, nil)
+
+        local conn_s1 = g.cluster:server('s1-master').net_box
+        local result = conn_s1.space[g.params.space_name]:get(known_bucket_id_key)
+        t.assert_equals(result, case.result.s1)
+
+        local conn_s2 = g.cluster:server('s2-master').net_box
+        local result = conn_s2.space[g.params.space_name]:get(known_bucket_id_key)
+        t.assert_equals(result, case.result.s2)
+    end
+end
+
+local known_bucket_id_read_cases = {
+    get = {
+        func = 'crud.get',
+        input_2 = known_bucket_id_key,
+        input_3 = {bucket_id = known_bucket_id},
+    },
+    select = {
+        func = 'crud.select',
+        input_2 = {{ '==', 'id', known_bucket_id_key}},
+        input_3 = {bucket_id = known_bucket_id},
+    },
+}
+
+for name, case in pairs(known_bucket_id_read_cases) do
+    local test_name = ('test_gh_278_%s_with_explicit_bucket_id_and_ddl'):format(name)
+
+    pgroup.before_test(test_name, prepare_known_bucket_id_data)
+
+    pgroup[test_name] = function(g)
+        local obj, err = g.cluster.main_server.net_box:call(
+            case.func, {
+                g.params.space_name,
+                case.input_2,
+                case.input_3,
+            })
+        t.assert_equals(err, nil)
+        t.assert_is_not(obj, nil)
+        t.assert_equals(obj.rows, {known_bucket_id_result_tuple})
+    end
+end
+
+pgroup.before_test(
+    'test_gh_278_pairs_with_explicit_bucket_id_and_ddl',
+    prepare_known_bucket_id_data)
+
+pgroup.test_gh_278_pairs_with_explicit_bucket_id_and_ddl = function(g)
+    local obj, err = g.cluster.main_server.net_box:eval([[
+        local res = {}
+        for _, row in crud.pairs(...) do
+            table.insert(res, row)
+        end
+
+        return res
+    ]], {
+        g.params.space_name,
+        {{ '==', 'id', known_bucket_id_key}},
+        {bucket_id = known_bucket_id}
+    })
+
+    t.assert_equals(err, nil)
+    t.assert_is_not(obj, nil)
+    t.assert_equals(obj, {known_bucket_id_result_tuple})
+end
+
+pgroup.before_test(
+    'test_gh_278_count_with_explicit_bucket_id_and_ddl',
+    prepare_known_bucket_id_data)
+
+pgroup.test_gh_278_count_with_explicit_bucket_id_and_ddl = function(g)
+    local obj, err = g.cluster.main_server.net_box:call(
+        'crud.count',
+        {
+            g.params.space_name,
+            {{ '==', 'id', known_bucket_id_key}},
+            {bucket_id = known_bucket_id}
+        })
+
+    t.assert_equals(err, nil)
+    t.assert_is_not(obj, nil)
+    t.assert_equals(obj, 1)
+end