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

91.3

/crud/update.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 sharding_key_module = require('crud.common.sharding.sharding_key')
local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
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 UpdateError = errors.new_class('UpdateError', {capture_stack = false})

local update = {}

local UPDATE_FUNC_NAME = 'update_on_storage'
local CRUD_UPDATE_FUNC_NAME = utils.get_storage_call(UPDATE_FUNC_NAME)

local function update_on_storage(space_name, key, operations, field_names, opts)
    dev_checks('string', '?', 'table', '?table', {
        bucket_id = 'number|cdata',
        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, UpdateError: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_update()
        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)
        if not ref_ok then
            return nil, bucket_ref_err
        end

        -- add_space_schema_hash is false because
        -- reloading space format on router can't avoid update error on storage
        local res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
            add_space_schema_hash = false,
            field_names = field_names,
            noreturn = opts.noreturn,
            fetch_latest_metadata = opts.fetch_latest_metadata,
        })

        if err == nil and res.err ~= nil and utils.is_field_not_found(res.err.code) then
            -- Relevant for Tarantool older than 2.8.1.
            -- We can only add fields to end of the tuple.
            -- If schema is updated and nullable fields are added, then we will get error.
            -- Therefore, we need to add filling of intermediate nullable fields.
            -- More details: https://github.com/tarantool/tarantool/issues/3378
            operations = utils.add_intermediate_nullable_fields(operations, space:format(), space:get(key))
            res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
                add_space_schema_hash = false,
                field_names = field_names,
                noreturn = opts.noreturn,
                fetch_latest_metadata = opts.fetch_latest_metadata,
            })
        end

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

        return res, err
    end

    return box.atomic(make_update)
end

update.storage_api = {[UPDATE_FUNC_NAME] = update_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_update_on_router(vshard_router, space_name, key, user_operations, opts)
    dev_checks('table', 'string', '?', 'table', {
        timeout = '?number',
        bucket_id = '?',
        fields = '?table',
        vshard_router = '?string|table',
        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, UpdateError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
    end
    if space == nil then
        return nil, UpdateError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
    end

    local space_format = space:format()

    if box.tuple.is(key) then
        key = key:totable()
    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_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)
        if err ~= nil then
            return nil, err
        end

        sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,
                                                                space_name,
                                                                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 operations = user_operations
    if not utils.tarantool_supports_fieldpaths() then
        operations, err = utils.convert_operations(user_operations, space_format)
        if err ~= nil then
            return nil, UpdateError:new("Wrong operations are specified: %s", err), const.NEED_SCHEMA_RELOAD
        end
    end

    local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)
    if err ~= nil then
        return nil, err
    end

    -- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
    sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)

    local update_on_storage_opts = {
        bucket_id = bucket_id_data.bucket_id,
        sharding_func_hash = bucket_id_data.sharding_func_hash,
        sharding_key_hash = sharding_key_hash,
        skip_sharding_hash_check = 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,
        bucket_id_data.bucket_id, CRUD_UPDATE_FUNC_NAME,
        {space_name, key, operations, opts.fields, update_on_storage_opts},
        call_opts
    )

    if err ~= nil then
        local err_wrapped = UpdateError:new("Failed to call update 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
        return nil, UpdateError:new("Failed to update: %s", storage_result.err)
    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

--- Updates tuple in the specified space
--
-- @function call
--
-- @param string space_name
--  A space name
--
-- @param key
--  Primary key value
--
-- @param table user_operations
--  Operations to be performed.
--  See `space:update` operations in Tarantool doc
--
-- @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] object
-- @treturn[2] nil
-- @treturn[2] table Error description
--
function update.call(space_name, key, user_operations, opts)
    checks('string', '?', 'table', {
        timeout = '?number',
        bucket_id = '?',
        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, UpdateError:new(err)
    end

    return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_update_on_router,
                                   space_name, key, user_operations, opts)
end

return update

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 sharding_key_module = require('crud.common.sharding.sharding_key')	258✔
9	local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')	258✔
10	local dev_checks = require('crud.common.dev_checks')	258✔
11	local schema = require('crud.common.schema')	258✔
12	local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')	258✔
13
14	local UpdateError = errors.new_class('UpdateError', {capture_stack = false})	258✔
15
16	local update = {}	258✔
17
18	local UPDATE_FUNC_NAME = 'update_on_storage'	258✔
19	local CRUD_UPDATE_FUNC_NAME = utils.get_storage_call(UPDATE_FUNC_NAME)	258✔
20
21	local function update_on_storage(space_name, key, operations, field_names, opts)
22	dev_checks('string', '?', 'table', '?table', {	66✔
23	bucket_id = 'number\|cdata',
24	sharding_key_hash = '?number',
25	sharding_func_hash = '?number',
26	skip_sharding_hash_check = '?boolean',
27	noreturn = '?boolean',
28	fetch_latest_metadata = '?boolean',
29	})
30
31	opts = opts or {}	66✔
32
33	local space = box.space[space_name]	66✔
34	if space == nil then	66✔
35	return nil, UpdateError:new("Space %q doesn't exist", space_name)	×
36	end
37
38	local _, err = sharding.check_sharding_hash(space_name,	132✔
39	opts.sharding_func_hash,	66✔
40	opts.sharding_key_hash,	66✔
41	opts.skip_sharding_hash_check)	66✔
42
43	if err ~= nil then	66✔
44	return nil, err	2✔
45	end
46
47	local function make_update()
48	local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)	64✔
49	if not ref_ok then	64✔
NEW 50	return nil, bucket_ref_err	×
51	end
52
53	-- add_space_schema_hash is false because
54	-- reloading space format on router can't avoid update error on storage
55	local res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {	128✔
56	add_space_schema_hash = false,
57	field_names = field_names,	64✔
58	noreturn = opts.noreturn,	64✔
59	fetch_latest_metadata = opts.fetch_latest_metadata,	64✔
60	})
61
62	if err == nil and res.err ~= nil and utils.is_field_not_found(res.err.code) then	82✔
63	-- Relevant for Tarantool older than 2.8.1.
64	-- We can only add fields to end of the tuple.
65	-- If schema is updated and nullable fields are added, then we will get error.
66	-- Therefore, we need to add filling of intermediate nullable fields.
67	-- More details: https://github.com/tarantool/tarantool/issues/3378
68	operations = utils.add_intermediate_nullable_fields(operations, space:format(), space:get(key))	4✔
69	res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {	2✔
70	add_space_schema_hash = false,
71	field_names = field_names,	1✔
72	noreturn = opts.noreturn,	1✔
73	fetch_latest_metadata = opts.fetch_latest_metadata,	1✔
74	})	1✔
75	end
76
77	local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(opts.bucket_id)	64✔
78	if not unref_ok then	64✔
NEW 79	return nil, err_unref	×
80	end
81
82	return res, err	64✔
83	end
84
85	return box.atomic(make_update)	64✔
86	end
87
88	update.storage_api = {[UPDATE_FUNC_NAME] = update_on_storage}	258✔
89
90	-- returns result, err, need_reload
91	-- need_reload indicates if reloading schema could help
92	-- see crud.common.schema.wrap_func_reload()
93	local function call_update_on_router(vshard_router, space_name, key, user_operations, opts)
94	dev_checks('table', 'string', '?', 'table', {	76✔
95	timeout = '?number',
96	bucket_id = '?',
97	fields = '?table',
98	vshard_router = '?string\|table',
99	noreturn = '?boolean',
100	fetch_latest_metadata = '?boolean',
101	})
102
103	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	76✔
104	if err ~= nil then	76✔
105	return nil, UpdateError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
106	end
107	if space == nil then	76✔
108	return nil, UpdateError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	16✔
109	end
110
111	local space_format = space:format()	68✔
112
113	if box.tuple.is(key) then	136✔
114	key = key:totable()	×
115	end
116
117	local sharding_key = key	68✔
118	local sharding_key_hash = nil	68✔
119	local skip_sharding_hash_check = nil
120
121	if opts.bucket_id == nil then	68✔
122	local primary_index_parts = space.index[0].parts	62✔
123
124	local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)	62✔
125	if err ~= nil then	62✔
126	return nil, err	×
127	end
128
129	sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,	124✔
130	space_name,	62✔
131	sharding_key_data.value,	62✔
132	primary_index_parts, key)	124✔
133	if err ~= nil then	62✔
134	return nil, err	2✔
135	end
136
137	sharding_key_hash = sharding_key_data.hash	60✔
138	else
139	skip_sharding_hash_check = true	6✔
140	end
141
142	local operations = user_operations	66✔
143	if not utils.tarantool_supports_fieldpaths() then	132✔
144	operations, err = utils.convert_operations(user_operations, space_format)	×
145	if err ~= nil then	×
146	return nil, UpdateError:new("Wrong operations are specified: %s", err), const.NEED_SCHEMA_RELOAD	×
147	end
148	end
149
150	local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)	66✔
151	if err ~= nil then	66✔
152	return nil, err	1✔
153	end
154
155	-- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
156	sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)	65✔
157
158	local update_on_storage_opts = {	65✔
159	bucket_id = bucket_id_data.bucket_id,	65✔
160	sharding_func_hash = bucket_id_data.sharding_func_hash,	65✔
161	sharding_key_hash = sharding_key_hash,	65✔
162	skip_sharding_hash_check = skip_sharding_hash_check,	65✔
163	noreturn = opts.noreturn,	65✔
164	fetch_latest_metadata = opts.fetch_latest_metadata,	65✔
165	}
166
167	local call_opts = {	65✔
168	mode = 'write',
169	timeout = opts.timeout,	65✔
170	}
171
172	local storage_result, err = call.single(vshard_router,	130✔
173	bucket_id_data.bucket_id, CRUD_UPDATE_FUNC_NAME,	65✔
174	{space_name, key, operations, opts.fields, update_on_storage_opts},	65✔
175	call_opts
176	)	65✔
177
178	if err ~= nil then	65✔
179	local err_wrapped = UpdateError:new("Failed to call update on storage-side: %s", err)	1✔
180
181	if sharding.result_needs_sharding_reload(err) then	2✔
182	return nil, err_wrapped, const.NEED_SHARDING_RELOAD	1✔
183	end
184
185	return nil, err_wrapped	×
186	end
187
188	if storage_result.err ~= nil then	64✔
189	return nil, UpdateError:new("Failed to update: %s", storage_result.err)	36✔
190	end
191
192	if opts.noreturn == true then	46✔
193	return nil	1✔
194	end
195
196	local tuple = storage_result.res	45✔
197
198	if opts.fetch_latest_metadata == true then	45✔
199	-- This option is temporary and is related to [1], [2].
200	-- [1] https://github.com/tarantool/crud/issues/236
201	-- [2] https://github.com/tarantool/crud/issues/361
202	space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,	4✔
203	vshard_router, opts, storage_result.storage_info)	4✔
204	end
205
206	return utils.format_result({tuple}, space, opts.fields)	45✔
207	end
208
209	--- Updates tuple in the specified space
210	--
211	-- @function call
212	--
213	-- @param string space_name
214	-- A space name
215	--
216	-- @param key
217	-- Primary key value
218	--
219	-- @param table user_operations
220	-- Operations to be performed.
221	-- See `space:update` operations in Tarantool doc
222	--
223	-- @tparam ?number opts.timeout
224	-- Function call timeout
225	--
226	-- @tparam ?number opts.bucket_id
227	-- Bucket ID
228	-- (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
229	--
230	-- @tparam ?string\|table opts.vshard_router
231	-- Cartridge vshard group name or vshard router instance.
232	-- Set this parameter if your space is not a part of the
233	-- default vshard cluster.
234	--
235	-- @tparam ?boolean opts.noreturn
236	-- Suppress returning successfully processed tuple.
237	--
238	-- @return[1] object
239	-- @treturn[2] nil
240	-- @treturn[2] table Error description
241	--
242	function update.call(space_name, key, user_operations, opts)	258✔
243	checks('string', '?', 'table', {	75✔
244	timeout = '?number',
245	bucket_id = '?',
246	fields = '?table',
247	vshard_router = '?string\|table',
248	noreturn = '?boolean',
249	fetch_latest_metadata = '?boolean',
250	})
251
252	opts = opts or {}	75✔
253	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	75✔
254	if err ~= nil then	75✔
255	return nil, UpdateError:new(err)	8✔
256	end
257
258	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_update_on_router,	71✔
259	space_name, key, user_operations, opts)	71✔
260	end
261
262	return update	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