8518810120

Committed 02 Apr 2024 07:09AM UTC coverage: 88.753% (+0.06%) from 88.694%

Build # 8518810120

Build Type

push

github

Committed by

DifferentialOrange

Commit Message

storage: use async bootstrap by default for tarantool 3

Closes #412
Part of #415

Run Details

8 of 9 new or added lines in 2 files covered. (88.89%)

209 existing lines in 16 files now uncovered.

4806 of 5415 relevant lines covered (88.75%)

6165.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.92

/crud/insert.lua

local checks = require('checks')
local errors = require('errors')

local call = require('crud.common.call')
local const = require('crud.common.const')
local utils = require('crud.common.utils')
local sharding = require('crud.common.sharding')
local dev_checks = require('crud.common.dev_checks')
local schema = require('crud.common.schema')

local InsertError = errors.new_class('InsertError', {capture_stack = false})

local insert = {}

local INSERT_FUNC_NAME = 'insert_on_storage'
local CRUD_INSERT_FUNC_NAME = utils.get_storage_call(INSERT_FUNC_NAME)

local function insert_on_storage(space_name, tuple, opts)
    dev_checks('string', 'table', {
        add_space_schema_hash = '?boolean',
        fields = '?table',
        sharding_key_hash = '?number',
        sharding_func_hash = '?number',
        skip_sharding_hash_check = '?boolean',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local space = box.space[space_name]
    if space == nil then
        return nil, InsertError: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 true only in case of insert_object
    -- the only one case when reloading schema can avoid insert error
    -- is flattening object on router
    return schema.wrap_box_space_func_result(space, 'insert', {tuple}, {
        add_space_schema_hash = opts.add_space_schema_hash,
        field_names = opts.fields,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    })
end

insert.storage_api = {[INSERT_FUNC_NAME] = insert_on_storage}

-- returns result, err, need_reload
-- need_reload indicates if reloading schema could help
-- see crud.common.schema.wrap_func_reload()
local function call_insert_on_router(vshard_router, space_name, original_tuple, opts)
    dev_checks('table', 'string', 'table', {
        timeout = '?number',
        bucket_id = '?number|cdata',
        add_space_schema_hash = '?boolean',
        fields = '?table',
        vshard_router = '?string|table',
        skip_nullability_check_on_flatten = '?boolean',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)
    if err ~= nil then
        return nil, InsertError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
    end
    if space == nil then
        return nil, InsertError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
    end

    local tuple = table.deepcopy(original_tuple)

    local sharding_data, err = sharding.tuple_set_and_return_bucket_id(vshard_router, tuple, space, opts.bucket_id)
    if err ~= nil then
        return nil, InsertError:new("Failed to get bucket ID: %s", err), const.NEED_SCHEMA_RELOAD
    end

    local insert_on_storage_opts = {
        add_space_schema_hash = opts.add_space_schema_hash,
        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,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    }

    local call_opts = {
        mode = 'write',
        timeout = opts.timeout,
    }

    local storage_result, err = call.single(vshard_router,
        sharding_data.bucket_id, CRUD_INSERT_FUNC_NAME,
        {space_name, tuple, insert_on_storage_opts},
        call_opts
    )

    if err ~= nil then
        local err_wrapped = InsertError:new("Failed to call insert on storage-side: %s", err)

        if sharding.result_needs_sharding_reload(err) then
            return nil, err_wrapped, const.NEED_SHARDING_RELOAD
        end

        return nil, err_wrapped
    end

    if storage_result.err ~= nil then
        local err_wrapped = InsertError:new("Failed to insert: %s", storage_result.err)

        if schema.result_needs_reload(space, storage_result) then
            return nil, err_wrapped, const.NEED_SCHEMA_RELOAD
        end

        return nil, err_wrapped
    end

    if opts.noreturn == true then
        return nil
    end

    local tuple = storage_result.res

    if opts.fetch_latest_metadata == true then
        -- This option is temporary and is related to [1], [2].
        -- [1] https://github.com/tarantool/crud/issues/236
        -- [2] https://github.com/tarantool/crud/issues/361
        space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,
                                                                vshard_router, opts, storage_result.storage_info)
    end

    return utils.format_result({tuple}, space, opts.fields)
end

--- Inserts a tuple to the specified space
--
-- @function tuple
--
-- @param string space_name
--  A space name
--
-- @param table tuple
--  Tuple
--
-- @tparam ?number opts.timeout
--  Function call timeout
--
-- @tparam ?number opts.bucket_id
--  Bucket ID
--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
--
-- @tparam ?string|table opts.vshard_router
--  Cartridge vshard group name or vshard router instance.
--  Set this parameter if your space is not a part of the
--  default vshard cluster.
--
-- @tparam ?boolean opts.noreturn
--  Suppress returning successfully processed tuple.
--
-- @return[1] tuple
-- @treturn[2] nil
-- @treturn[2] table Error description
--
function insert.tuple(space_name, tuple, opts)
    checks('string', 'table', {
        timeout = '?number',
        bucket_id = '?number|cdata',
        add_space_schema_hash = '?boolean',
        fields = '?table',
        vshard_router = '?string|table',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)
    if err ~= nil then
        return nil, InsertError:new(err)
    end

    return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_insert_on_router,
                                   space_name, tuple, opts)
end

--- Inserts an object to the specified space
--
-- @function object
--
-- @param string space_name
--  A space name
--
-- @param table obj
--  Object
--
-- @tparam ?table opts
--  Options of insert.tuple
--
-- @return[1] object
-- @treturn[2] nil
-- @treturn[2] table Error description
--
function insert.object(space_name, obj, opts)
    checks('string', 'table', {
        timeout = '?number',
        bucket_id = '?number|cdata',
        add_space_schema_hash = '?boolean',
        fields = '?table',
        vshard_router = '?string|table',
        skip_nullability_check_on_flatten = '?boolean',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)
    if err ~= nil then
        return nil, InsertError:new(err)
    end

    -- insert can fail if router uses outdated schema to flatten object
    opts = utils.merge_options(opts, {add_space_schema_hash = true})

    local tuple, err = utils.flatten_obj_reload(vshard_router, space_name, obj,
                                                opts.skip_nullability_check_on_flatten)
    if err ~= nil then
        return nil, InsertError:new("Failed to flatten object: %s", err)
    end

    return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_insert_on_router,
                                   space_name, tuple, opts)
end

return insert

1	local checks = require('checks')	248✔
2	local errors = require('errors')	248✔
3
4	local call = require('crud.common.call')	248✔
5	local const = require('crud.common.const')	248✔
6	local utils = require('crud.common.utils')	248✔
7	local sharding = require('crud.common.sharding')	248✔
8	local dev_checks = require('crud.common.dev_checks')	248✔
9	local schema = require('crud.common.schema')	248✔
10
11	local InsertError = errors.new_class('InsertError', {capture_stack = false})	248✔
12
13	local insert = {}	248✔
14
15	local INSERT_FUNC_NAME = 'insert_on_storage'	248✔
16	local CRUD_INSERT_FUNC_NAME = utils.get_storage_call(INSERT_FUNC_NAME)	248✔
17
18	local function insert_on_storage(space_name, tuple, opts)
19	dev_checks('string', 'table', {	64,352✔
20	add_space_schema_hash = '?boolean',
21	fields = '?table',
22	sharding_key_hash = '?number',
23	sharding_func_hash = '?number',
24	skip_sharding_hash_check = '?boolean',
25	noreturn = '?boolean',
26	fetch_latest_metadata = '?boolean',
27	})
28
29	opts = opts or {}	64,352✔
30
31	local space = box.space[space_name]	64,352✔
32	if space == nil then	64,352✔
33	return nil, InsertError:new("Space %q doesn't exist", space_name)	56✔
34	end
35
36	local _, err = sharding.check_sharding_hash(space_name,	128,648✔
37	opts.sharding_func_hash,	64,324✔
38	opts.sharding_key_hash,	64,324✔
39	opts.skip_sharding_hash_check)	64,324✔
40
41	if err ~= nil then	64,324✔
42	return nil, err	72✔
43	end
44
45	-- add_space_schema_hash is true only in case of insert_object
46	-- the only one case when reloading schema can avoid insert error
47	-- is flattening object on router
48	return schema.wrap_box_space_func_result(space, 'insert', {tuple}, {	64,252✔
49	add_space_schema_hash = opts.add_space_schema_hash,	64,252✔
50	field_names = opts.fields,	64,252✔
51	noreturn = opts.noreturn,	64,252✔
52	fetch_latest_metadata = opts.fetch_latest_metadata,	64,252✔
53	})	64,252✔
54	end
55
56	insert.storage_api = {[INSERT_FUNC_NAME] = insert_on_storage}	248✔
57
58	-- returns result, err, need_reload
59	-- need_reload indicates if reloading schema could help
60	-- see crud.common.schema.wrap_func_reload()
61	local function call_insert_on_router(vshard_router, space_name, original_tuple, opts)
62	dev_checks('table', 'string', 'table', {	64,333✔
63	timeout = '?number',
64	bucket_id = '?number\|cdata',
65	add_space_schema_hash = '?boolean',
66	fields = '?table',
67	vshard_router = '?string\|table',
68	skip_nullability_check_on_flatten = '?boolean',
69	noreturn = '?boolean',
70	fetch_latest_metadata = '?boolean',
71	})
72
73	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	64,333✔
74	if err ~= nil then	64,333✔
UNCOV 75	return nil, InsertError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
76	end
77	if space == nil then	64,333✔
78	return nil, InsertError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	44✔
79	end
80
81	local tuple = table.deepcopy(original_tuple)	64,311✔
82
83	local sharding_data, err = sharding.tuple_set_and_return_bucket_id(vshard_router, tuple, space, opts.bucket_id)	64,311✔
84	if err ~= nil then	64,311✔
85	return nil, InsertError:new("Failed to get bucket ID: %s", err), const.NEED_SCHEMA_RELOAD	18✔
86	end
87
88	local insert_on_storage_opts = {	64,302✔
89	add_space_schema_hash = opts.add_space_schema_hash,	64,302✔
90	fields = opts.fields,	64,302✔
91	sharding_func_hash = sharding_data.sharding_func_hash,	64,302✔
92	sharding_key_hash = sharding_data.sharding_key_hash,	64,302✔
93	skip_sharding_hash_check = sharding_data.skip_sharding_hash_check,	64,302✔
94	noreturn = opts.noreturn,	64,302✔
95	fetch_latest_metadata = opts.fetch_latest_metadata,	64,302✔
96	}
97
98	local call_opts = {	64,302✔
99	mode = 'write',
100	timeout = opts.timeout,	64,302✔
101	}
102
103	local storage_result, err = call.single(vshard_router,	128,604✔
104	sharding_data.bucket_id, CRUD_INSERT_FUNC_NAME,	64,302✔
105	{space_name, tuple, insert_on_storage_opts},	64,302✔
106	call_opts
107	)	64,302✔
108
109	if err ~= nil then	64,302✔
110	local err_wrapped = InsertError:new("Failed to call insert on storage-side: %s", err)	50✔
111
112	if sharding.result_needs_sharding_reload(err) then	100✔
113	return nil, err_wrapped, const.NEED_SHARDING_RELOAD	36✔
114	end
115
116	return nil, err_wrapped	14✔
117	end
118
119	if storage_result.err ~= nil then	64,252✔
120	local err_wrapped = InsertError:new("Failed to insert: %s", storage_result.err)	624✔
121
122	if schema.result_needs_reload(space, storage_result) then	1,248✔
123	return nil, err_wrapped, const.NEED_SCHEMA_RELOAD	6✔
124	end
125
126	return nil, err_wrapped	618✔
127	end
128
129	if opts.noreturn == true then	63,628✔
130	return nil	2✔
131	end
132
133	local tuple = storage_result.res	63,626✔
134
135	if opts.fetch_latest_metadata == true then	63,626✔
136	-- This option is temporary and is related to [1], [2].
137	-- [1] https://github.com/tarantool/crud/issues/236
138	-- [2] https://github.com/tarantool/crud/issues/361
139	space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,	4✔
140	vshard_router, opts, storage_result.storage_info)	4✔
141	end
142
143	return utils.format_result({tuple}, space, opts.fields)	63,626✔
144	end
145
146	--- Inserts a tuple to the specified space
147	--
148	-- @function tuple
149	--
150	-- @param string space_name
151	-- A space name
152	--
153	-- @param table tuple
154	-- Tuple
155	--
156	-- @tparam ?number opts.timeout
157	-- Function call timeout
158	--
159	-- @tparam ?number opts.bucket_id
160	-- Bucket ID
161	-- (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
162	--
163	-- @tparam ?string\|table opts.vshard_router
164	-- Cartridge vshard group name or vshard router instance.
165	-- Set this parameter if your space is not a part of the
166	-- default vshard cluster.
167	--
168	-- @tparam ?boolean opts.noreturn
169	-- Suppress returning successfully processed tuple.
170	--
171	-- @return[1] tuple
172	-- @treturn[2] nil
173	-- @treturn[2] table Error description
174	--
175	function insert.tuple(space_name, tuple, opts)	248✔
176	checks('string', 'table', {	60,369✔
177	timeout = '?number',
178	bucket_id = '?number\|cdata',
179	add_space_schema_hash = '?boolean',
180	fields = '?table',
181	vshard_router = '?string\|table',
182	noreturn = '?boolean',
183	fetch_latest_metadata = '?boolean',
184	})
185
186	opts = opts or {}	60,369✔
187
188	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	60,369✔
189	if err ~= nil then	60,369✔
190	return nil, InsertError:new(err)	8✔
191	end
192
193	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_insert_on_router,	60,365✔
194	space_name, tuple, opts)	60,365✔
195	end
196
197	--- Inserts an object to the specified space
198	--
199	-- @function object
200	--
201	-- @param string space_name
202	-- A space name
203	--
204	-- @param table obj
205	-- Object
206	--
207	-- @tparam ?table opts
208	-- Options of insert.tuple
209	--
210	-- @return[1] object
211	-- @treturn[2] nil
212	-- @treturn[2] table Error description
213	--
214	function insert.object(space_name, obj, opts)	248✔
215	checks('string', 'table', {	3,932✔
216	timeout = '?number',
217	bucket_id = '?number\|cdata',
218	add_space_schema_hash = '?boolean',
219	fields = '?table',
220	vshard_router = '?string\|table',
221	skip_nullability_check_on_flatten = '?boolean',
222	noreturn = '?boolean',
223	fetch_latest_metadata = '?boolean',
224	})
225
226	opts = opts or {}	3,932✔
227
228	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	3,932✔
229	if err ~= nil then	3,932✔
230	return nil, InsertError:new(err)	8✔
231	end
232
233	-- insert can fail if router uses outdated schema to flatten object
234	opts = utils.merge_options(opts, {add_space_schema_hash = true})	7,856✔
235
236	local tuple, err = utils.flatten_obj_reload(vshard_router, space_name, obj,	7,856✔
237	opts.skip_nullability_check_on_flatten)	3,928✔
238	if err ~= nil then	3,928✔
239	return nil, InsertError:new("Failed to flatten object: %s", err)	42✔
240	end
241
242	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_insert_on_router,	3,907✔
243	space_name, tuple, opts)	3,907✔
244	end
245
246	return insert	248✔

tarantool / crud / 8518810120

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous