20723023882

Committed 05 Jan 2026 05:07PM UTC coverage: 88.782% (+0.4%) from 88.405%

Build # 20723023882

Build Type

Pull #475

github

Committed by

Satbek

Commit Message

:try with decorator

Pull Request Pull Request #475: safe/fast mode: remove fiber kill

Run Details

1335 of 1428 new or added lines in 12 files covered. (93.49%)

74 existing lines in 10 files now uncovered.

5334 of 6008 relevant lines covered (88.78%)

13774.25 hits per line

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

97.2

/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 bucket_ref_unref = require("crud.common.sharding.bucket_ref_unref")

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

        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
        local ref_ok, bucket_ref_err, unref = bucket_ref_unref.bucket_refrw(bucket_id)

        if not ref_ok then
                return nil, bucket_ref_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
        local result = schema.wrap_func_result(space, space.insert, {
                add_space_schema_hash = opts.add_space_schema_hash,
                field_names = opts.fields,
                noreturn = opts.noreturn,
                fetch_latest_metadata = opts.fetch_latest_metadata,
        }, space, tuple)

        local unref_ok, err_unref = unref(bucket_id)
        if not unref_ok then
                return nil, err_unref
        end

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

tarantool / crud / 20723023882

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