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

81.73

/crud/borders.lua

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

local const = require('crud.common.const')
local dev_checks = require('crud.common.dev_checks')
local call = require('crud.common.call')
local utils = require('crud.common.utils')
local schema = require('crud.common.schema')
local has_keydef, Keydef = pcall(require, 'crud.compare.keydef')
local select_comparators = require('crud.compare.comparators')

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

local borders = {}

local STAT_FUNC_NAME = 'get_border_on_storage'
local CRUD_STAT_FUNC_NAME = utils.get_storage_call(STAT_FUNC_NAME)


local function get_border_on_storage(border_name, space_name, index_id, field_names, fetch_latest_metadata)
    dev_checks('string', 'string', 'number', '?table', '?boolean')

    assert(border_name == 'min' or border_name == 'max')

    local space = box.space[space_name]
    if space == nil then
        return nil, BorderError:new("Space %q doesn't exist", space_name)
    end

    local index = space.index[index_id]
    if index == nil then
        return nil, BorderError:new("Index %q of space doesn't exist", index_id, space_name)
    end

    local function get_index_border(index)
        return index[border_name](index)
    end

    return schema.wrap_func_result(space, get_index_border, {index}, {
        add_space_schema_hash = true,
        field_names = field_names,
        fetch_latest_metadata = fetch_latest_metadata,
    })
end

borders.storage_api = {[STAT_FUNC_NAME] = get_border_on_storage}

local is_closer

if has_keydef then
    is_closer = function (compare_sign, keydef, tuple, res_tuple)
        if res_tuple == nil then
            return true
        end

        local cmp = keydef:compare(tuple, res_tuple)

        return cmp * compare_sign > 0
    end
else
    is_closer = function (_, comparator, tuple, res_tuple)
        if res_tuple == nil then
            return true
        end
        return comparator(tuple, res_tuple)
    end
end

local function call_get_border_on_router(vshard_router, border_name, space_name, index_name, opts)
    checks('table', 'string', 'string', '?string|number', {
        timeout = '?number',
        fields = '?table',
        mode = '?string',
        vshard_router = '?string|table',
        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, BorderError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
    end
    if space == nil then
        return nil, BorderError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
    end

    local index
    if index_name == nil then
        index = space.index[0]
    else
        index = space.index[index_name]
    end

    if index == nil then
        return nil,
               BorderError:new("Index %q of space %q doesn't exist", index_name, space_name),
               const.NEED_SCHEMA_RELOAD
    end

    local primary_index = space.index[0]

    local cmp_key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)
    local field_names = utils.enrich_field_names_with_cmp_key(opts.fields, cmp_key_parts, space:format())

    local replicasets, err = vshard_router:routeall()
    if err ~= nil then
        return nil, BorderError:new("Failed to get router replicasets: %s", err)
    end
    local call_opts = {
        mode = opts.mode or 'read',
        replicasets = replicasets,
        timeout = opts.timeout,
    }
    local results, err, storages_info = call.map(vshard_router,
        CRUD_STAT_FUNC_NAME,
        {border_name, space_name, index.id, field_names, opts.fetch_latest_metadata},
        call_opts
    )

    if err ~= nil then
        return nil, BorderError:new("Failed to get %s: %s", border_name, err)
    end

    local compare_sign = border_name == 'max' and 1 or -1
    local comparator
    if has_keydef then
        comparator = Keydef.new(space, field_names, index.id)
    else
        local tarantool_iter
        if compare_sign > 0 then
            tarantool_iter = box.index.GT
        else
            tarantool_iter = box.index.LT
        end
        local key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)
        local cmp_operator = select_comparators.get_cmp_operator(tarantool_iter)
        comparator = select_comparators.gen_tuples_comparator(cmp_operator, key_parts, field_names, space:format())
    end

    local res_tuple = nil
    for _, storage_result in pairs(results) do
        local storage_result = storage_result[1]
        if storage_result.err ~= nil then
            local err_wrapped = BorderError:new("Failed to get %s: %s", border_name, storage_result.err)

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

            return nil, err_wrapped
        end

        local tuple = storage_result.res
        if tuple ~= nil and is_closer(compare_sign, comparator, tuple, res_tuple) then
            res_tuple = tuple
        end
    end

    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_map_storages(space, space_name, vshard_router, opts,
                                                              storages_info, netbox_schema_version)
    end

    local result = utils.format_result({res_tuple}, space, field_names)

    if opts.fields ~= nil then
        result = utils.cut_rows(result.rows, result.metadata, opts.fields)
    end

    return result
end

local function get_border(border_name, space_name, index_name, opts)
    opts = opts or {}
    local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)
    if err ~= nil then
        return nil, BorderError:new(err)
    end

    return schema.wrap_func_reload(vshard_router, call_get_border_on_router,
        border_name, space_name, index_name, opts
    )
end

--- Find the minimum value in the specified index
--
-- @function min
--
-- @param string space_name
--  A space name
--
-- @param ?string index_name
--  An index name (by default, primary index is used)
--
-- @tparam ?number opts.timeout
--  Function call timeout
--
-- @tparam ?table opts.fields
--  Field names for getting only a subset of fields
--
-- @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.
--
-- @return[1] result
-- @treturn[2] nil
-- @treturn[2] table Error description
function borders.min(space_name, index_id, opts)
    return get_border('min', space_name, index_id, opts)
end

--- Find the maximum value in the specified index
--
-- @function min
--
-- @param string space_name
--  A space name
--
-- @param ?string index_name
--  An index name (by default, primary index is used)
--
-- @tparam ?number opts.timeout
--  Function call timeout
--
-- @tparam ?table opts.fields
--  Field names for getting only a subset of fields
--
-- @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.
--
-- @return[1] result
-- @treturn[2] nil
-- @treturn[2] table Error description
function borders.max(space_name, index_id, opts)
    return get_border('max', space_name, index_id, opts)
end

return borders

1	local checks = require('checks')	248✔
2	local errors = require('errors')	248✔
3
4	local const = require('crud.common.const')	248✔
5	local dev_checks = require('crud.common.dev_checks')	248✔
6	local call = require('crud.common.call')	248✔
7	local utils = require('crud.common.utils')	248✔
8	local schema = require('crud.common.schema')	248✔
9	local has_keydef, Keydef = pcall(require, 'crud.compare.keydef')	248✔
10	local select_comparators = require('crud.compare.comparators')	248✔
11
12	local BorderError = errors.new_class('BorderError', {capture_stack = false})	248✔
13
14	local borders = {}	248✔
15
16	local STAT_FUNC_NAME = 'get_border_on_storage'	248✔
17	local CRUD_STAT_FUNC_NAME = utils.get_storage_call(STAT_FUNC_NAME)	248✔
18
19
20	local function get_border_on_storage(border_name, space_name, index_id, field_names, fetch_latest_metadata)
21	dev_checks('string', 'string', 'number', '?table', '?boolean')	132✔
22
23	assert(border_name == 'min' or border_name == 'max')	132✔
24
25	local space = box.space[space_name]	132✔
26	if space == nil then	132✔
27	return nil, BorderError:new("Space %q doesn't exist", space_name)	×
28	end
29
30	local index = space.index[index_id]	132✔
31	if index == nil then	132✔
32	return nil, BorderError:new("Index %q of space doesn't exist", index_id, space_name)	×
33	end
34
35	local function get_index_border(index)
36	return index[border_name](index)	132✔
37	end
38
39	return schema.wrap_func_result(space, get_index_border, {index}, {	132✔
40	add_space_schema_hash = true,
41	field_names = field_names,	132✔
42	fetch_latest_metadata = fetch_latest_metadata,	132✔
43	})	132✔
44	end
45
46	borders.storage_api = {[STAT_FUNC_NAME] = get_border_on_storage}	248✔
47
48	local is_closer
49
50	if has_keydef then	248✔
51	is_closer = function (compare_sign, keydef, tuple, res_tuple)
52	if res_tuple == nil then	54✔
53	return true	28✔
54	end
55
56	local cmp = keydef:compare(tuple, res_tuple)	52✔
57
58	return cmp * compare_sign > 0	26✔
59	end
60	else
61	is_closer = function (_, comparator, tuple, res_tuple)
UNCOV 62	if res_tuple == nil then	×
UNCOV 63	return true	×
64	end
65	return comparator(tuple, res_tuple)	×
66	end
67	end
68
69	local function call_get_border_on_router(vshard_router, border_name, space_name, index_name, opts)
70	checks('table', 'string', 'string', '?string\|number', {	156✔
71	timeout = '?number',
72	fields = '?table',
73	mode = '?string',
74	vshard_router = '?string\|table',
75	fetch_latest_metadata = '?boolean',
76	})
77
78	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	156✔
79	if err ~= nil then	156✔
UNCOV 80	return nil, BorderError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
81	end
82	if space == nil then	156✔
83	return nil, BorderError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	36✔
84	end
85
86	local index
87	if index_name == nil then	138✔
88	index = space.index[0]	40✔
89	else
90	index = space.index[index_name]	98✔
91	end
92
93	if index == nil then	138✔
94	return nil,	72✔
95	BorderError:new("Index %q of space %q doesn't exist", index_name, space_name),	72✔
96	const.NEED_SCHEMA_RELOAD	72✔
97	end
98
99	local primary_index = space.index[0]	66✔
100
101	local cmp_key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)	66✔
102	local field_names = utils.enrich_field_names_with_cmp_key(opts.fields, cmp_key_parts, space:format())	132✔
103
104	local replicasets, err = vshard_router:routeall()	66✔
105	if err ~= nil then	66✔
UNCOV 106	return nil, BorderError:new("Failed to get router replicasets: %s", err)	×
107	end
108	local call_opts = {	66✔
109	mode = opts.mode or 'read',	66✔
110	replicasets = replicasets,	66✔
111	timeout = opts.timeout,	66✔
112	}
113	local results, err, storages_info = call.map(vshard_router,	132✔
114	CRUD_STAT_FUNC_NAME,	66✔
115	{border_name, space_name, index.id, field_names, opts.fetch_latest_metadata},	66✔
116	call_opts
117	)	66✔
118
119	if err ~= nil then	66✔
UNCOV 120	return nil, BorderError:new("Failed to get %s: %s", border_name, err)	×
121	end
122
123	local compare_sign = border_name == 'max' and 1 or -1	66✔
124	local comparator
125	if has_keydef then	66✔
126	comparator = Keydef.new(space, field_names, index.id)	132✔
127	else
128	local tarantool_iter
UNCOV 129	if compare_sign > 0 then	×
UNCOV 130	tarantool_iter = box.index.GT	×
131	else
132	tarantool_iter = box.index.LT	×
133	end
134	local key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)	×
UNCOV 135	local cmp_operator = select_comparators.get_cmp_operator(tarantool_iter)	×
136	comparator = select_comparators.gen_tuples_comparator(cmp_operator, key_parts, field_names, space:format())	×
137	end
138
139	local res_tuple = nil	66✔
140	for _, storage_result in pairs(results) do	264✔
141	local storage_result = storage_result[1]	132✔
142	if storage_result.err ~= nil then	132✔
UNCOV 143	local err_wrapped = BorderError:new("Failed to get %s: %s", border_name, storage_result.err)	×
144
145	local need_reload = schema.result_needs_reload(space, storage_result)	×
UNCOV 146	if need_reload then	×
147	return nil, err_wrapped, const.NEED_SCHEMA_RELOAD	×
148	end
149
UNCOV 150	return nil, err_wrapped	×
151	end
152
153	local tuple = storage_result.res	132✔
154	if tuple ~= nil and is_closer(compare_sign, comparator, tuple, res_tuple) then	186✔
155	res_tuple = tuple	34✔
156	end
157	end
158
159	if opts.fetch_latest_metadata == true then	66✔
160	-- This option is temporary and is related to [1], [2].
161	-- [1] https://github.com/tarantool/crud/issues/236
162	-- [2] https://github.com/tarantool/crud/issues/361
163	space = utils.fetch_latest_metadata_when_map_storages(space, space_name, vshard_router, opts,	4✔
164	storages_info, netbox_schema_version)	4✔
165	end
166
167	local result = utils.format_result({res_tuple}, space, field_names)	66✔
168
169	if opts.fields ~= nil then	66✔
170	result = utils.cut_rows(result.rows, result.metadata, opts.fields)	20✔
171	end
172
173	return result	66✔
174	end
175
176	local function get_border(border_name, space_name, index_name, opts)
177	opts = opts or {}	118✔
178	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	118✔
179	if err ~= nil then	118✔
180	return nil, BorderError:new(err)	16✔
181	end
182
183	return schema.wrap_func_reload(vshard_router, call_get_border_on_router,	110✔
184	border_name, space_name, index_name, opts	110✔
185	)	110✔
186	end
187
188	--- Find the minimum value in the specified index
189	--
190	-- @function min
191	--
192	-- @param string space_name
193	-- A space name
194	--
195	-- @param ?string index_name
196	-- An index name (by default, primary index is used)
197	--
198	-- @tparam ?number opts.timeout
199	-- Function call timeout
200	--
201	-- @tparam ?table opts.fields
202	-- Field names for getting only a subset of fields
203	--
204	-- @tparam ?string\|table opts.vshard_router
205	-- Cartridge vshard group name or vshard router instance.
206	-- Set this parameter if your space is not a part of the
207	-- default vshard cluster.
208	--
209	-- @return[1] result
210	-- @treturn[2] nil
211	-- @treturn[2] table Error description
212	function borders.min(space_name, index_id, opts)	248✔
213	return get_border('min', space_name, index_id, opts)	59✔
214	end
215
216	--- Find the maximum value in the specified index
217	--
218	-- @function min
219	--
220	-- @param string space_name
221	-- A space name
222	--
223	-- @param ?string index_name
224	-- An index name (by default, primary index is used)
225	--
226	-- @tparam ?number opts.timeout
227	-- Function call timeout
228	--
229	-- @tparam ?table opts.fields
230	-- Field names for getting only a subset of fields
231	--
232	-- @tparam ?string\|table opts.vshard_router
233	-- Cartridge vshard group name or vshard router instance.
234	-- Set this parameter if your space is not a part of the
235	-- default vshard cluster.
236	--
237	-- @return[1] result
238	-- @treturn[2] nil
239	-- @treturn[2] table Error description
240	function borders.max(space_name, index_id, opts)	248✔
241	return get_border('max', space_name, index_id, opts)	59✔
242	end
243
244	return borders	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