19762722361

Committed 28 Nov 2025 11:37AM UTC coverage: 88.331% (-0.1%) from 88.469%

Build # 19762722361

Build Type

Pull #463

github

Committed by

Satbek

Commit Message

change slow to safe in bucket_ref_unref module naming

Pull Request Pull Request #463: TNTP-2109: call bucket_ref/bucket_unref on crud operations

Run Details

205 of 268 new or added lines in 16 files covered. (76.49%)

10 existing lines in 1 file now uncovered.

5193 of 5879 relevant lines covered (88.33%)

6428.43 hits per line

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

97.06

/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 function make_insert()
        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
        local ref_ok, bucket_ref_err = 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_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,
        })

        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
        if not unref_ok then
            return nil, err_unref
        end

        return result
    end

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

tarantool / crud / 19762722361

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