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

92.55

/crud/delete.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 DeleteError = errors.new_class('DeleteError', {capture_stack = false})

local delete = {}

local DELETE_FUNC_NAME = 'delete_on_storage'
local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)

local function delete_on_storage(space_name, key, field_names, opts)
    dev_checks('string', '?', '?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, DeleteError: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 false because
    -- reloading space format on router can't avoid delete error on storage
    return schema.wrap_box_space_func_result(space, 'delete', {key}, {
        add_space_schema_hash = false,
        field_names = field_names,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    })
end

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

    if box.tuple.is(key) then
        key = key:totable()
    end

    local sharding_key_hash = nil
    local skip_sharding_hash_check = nil

    local sharding_key = key
    if opts.bucket_id == nil then
        if space.index[0] == nil then
            return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD
        end
        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 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

    local delete_on_storage_opts = {
        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_DELETE_FUNC_NAME,
        {space_name, key, opts.fields, delete_on_storage_opts},
        call_opts
    )

    if err ~= nil then
        local err_wrapped = DeleteError:new("Failed to call delete 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, DeleteError:new("Failed to delete: %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

--- Deletes tuple from the specified space by key
--
-- @function call
--
-- @param string space_name
--  A space name
--
-- @param key
--  Primary key value
--
-- @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 delete.call(space_name, key, opts)
    checks('string', '?', {
        timeout = '?number',
        bucket_id = '?number|cdata',
        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, DeleteError:new(err)
    end

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

return delete

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 sharding_key_module = require('crud.common.sharding.sharding_key')	248✔
9	local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')	248✔
10	local dev_checks = require('crud.common.dev_checks')	248✔
11	local schema = require('crud.common.schema')	248✔
12
13	local DeleteError = errors.new_class('DeleteError', {capture_stack = false})	248✔
14
15	local delete = {}	248✔
16
17	local DELETE_FUNC_NAME = 'delete_on_storage'	248✔
18	local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)	248✔
19
20	local function delete_on_storage(space_name, key, field_names, opts)
21	dev_checks('string', '?', '?table', {	75✔
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 {}	75✔
30
31	local space = box.space[space_name]	75✔
32	if space == nil then	75✔
33	return nil, DeleteError:new("Space %q doesn't exist", space_name)	×
34	end
35
36	local _, err = sharding.check_sharding_hash(space_name,	150✔
37	opts.sharding_func_hash,	75✔
38	opts.sharding_key_hash,	75✔
39	opts.skip_sharding_hash_check)	75✔
40
41	if err ~= nil then	75✔
42	return nil, err	2✔
43	end
44
45	-- add_space_schema_hash is false because
46	-- reloading space format on router can't avoid delete error on storage
47	return schema.wrap_box_space_func_result(space, 'delete', {key}, {	73✔
48	add_space_schema_hash = false,
49	field_names = field_names,	73✔
50	noreturn = opts.noreturn,	73✔
51	fetch_latest_metadata = opts.fetch_latest_metadata,	73✔
52	})	73✔
53	end
54
55	delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}	248✔
56
57	-- returns result, err, need_reload
58	-- need_reload indicates if reloading schema could help
59	-- see crud.common.schema.wrap_func_reload()
60	local function call_delete_on_router(vshard_router, space_name, key, opts)
61	dev_checks('table', 'string', '?', {	86✔
62	timeout = '?number',
63	bucket_id = '?number\|cdata',
64	fields = '?table',
65	vshard_router = '?string\|table',
66	noreturn = '?boolean',
67	fetch_latest_metadata = '?boolean',
68	})
69
70	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	86✔
71	if err ~= nil then	86✔
UNCOV 72	return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
73	end
74	if space == nil then	86✔
75	return nil, DeleteError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	20✔
76	end
77
78	if box.tuple.is(key) then	152✔
UNCOV 79	key = key:totable()	×
80	end
81
82	local sharding_key_hash = nil	76✔
83	local skip_sharding_hash_check = nil
84
85	local sharding_key = key	76✔
86	if opts.bucket_id == nil then	76✔
87	if space.index[0] == nil then	71✔
UNCOV 88	return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD	×
89	end
90	local primary_index_parts = space.index[0].parts	71✔
91
92	local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)	71✔
93	if err ~= nil then	71✔
UNCOV 94	return nil, err	×
95	end
96
97	sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,	142✔
98	space_name,	71✔
99	sharding_key_data.value,	71✔
100	primary_index_parts, key)	142✔
101	if err ~= nil then	71✔
102	return nil, err	2✔
103	end
104
105	sharding_key_hash = sharding_key_data.hash	69✔
106	else
107	skip_sharding_hash_check = true	5✔
108	end
109
110	local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)	74✔
111	if err ~= nil then	74✔
UNCOV 112	return nil, err	×
113	end
114
115	local delete_on_storage_opts = {	74✔
116	sharding_func_hash = bucket_id_data.sharding_func_hash,	74✔
117	sharding_key_hash = sharding_key_hash,	74✔
118	skip_sharding_hash_check = skip_sharding_hash_check,	74✔
119	noreturn = opts.noreturn,	74✔
120	fetch_latest_metadata = opts.fetch_latest_metadata,	74✔
121	}
122
123	local call_opts = {	74✔
124	mode = 'write',
125	timeout = opts.timeout,	74✔
126	}
127
128	local storage_result, err = call.single(vshard_router,	148✔
129	bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,	74✔
130	{space_name, key, opts.fields, delete_on_storage_opts},	74✔
131	call_opts
132	)	74✔
133
134	if err ~= nil then	74✔
135	local err_wrapped = DeleteError:new("Failed to call delete on storage-side: %s", err)	1✔
136
137	if sharding.result_needs_sharding_reload(err) then	2✔
138	return nil, err_wrapped, const.NEED_SHARDING_RELOAD	1✔
139	end
140
UNCOV 141	return nil, err_wrapped	×
142	end
143
144	if storage_result.err ~= nil then	73✔
145	return nil, DeleteError:new("Failed to delete: %s", storage_result.err)	34✔
146	end
147
148	if opts.noreturn == true then	56✔
149	return nil	1✔
150	end
151
152	local tuple = storage_result.res	55✔
153
154	if opts.fetch_latest_metadata == true then	55✔
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,	2✔
159	vshard_router, opts, storage_result.storage_info)	2✔
160	end
161
162	return utils.format_result({tuple}, space, opts.fields)	55✔
163	end
164
165	--- Deletes tuple from the specified space by key
166	--
167	-- @function call
168	--
169	-- @param string space_name
170	-- A space name
171	--
172	-- @param key
173	-- Primary key value
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] object
191	-- @treturn[2] nil
192	-- @treturn[2] table Error description
193	--
194	function delete.call(space_name, key, opts)	248✔
195	checks('string', '?', {	83✔
196	timeout = '?number',
197	bucket_id = '?number\|cdata',
198	fields = '?table',
199	vshard_router = '?string\|table',
200	noreturn = '?boolean',
201	fetch_latest_metadata = '?boolean',
202	})
203
204	opts = opts or {}	83✔
205
206	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	83✔
207	if err ~= nil then	83✔
208	return nil, DeleteError:new(err)	8✔
209	end
210
211	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_delete_on_router,	79✔
212	space_name, key, opts)	79✔
213	end
214
215	return delete	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