5199459215

pending completion

Build # 5199459215

Build Type

push

github

Committed by

DifferentialOrange

Commit Message

Release 1.2.0

Overview

  This release add two new flags: `noreturn` to ignore return values
  excessive transfer and encoding/decoding for insert/replace/etc
  (performance improvement up to 10% for batch requests) and
  `fetch_latest_metadata` to force fetching latest space format metadata
  right after a live migration (performance overhead may be up to 15%).

New features
  * Add `noreturn` option for operations:
    `insert`, `insert_object`, `insert_many`, `insert_object_many`,
    `replace`, `replace_object`, `replace_many`, `insert_object_many`,
    `upsert`, `upsert_object`, `upsert_many`, `upsert_object_many`,
    `update`, `delete` (#267).

Bugfixes
  * Crud DML operations returning stale schema for metadata generation.
    Now you may use `fetch_latest_metadata` flag to work with latest
    schema (#236).

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

4549 of 4888 relevant lines covered (93.06%)

18261.17 hits per line

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

81.55

/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 = '_crud.get_border_on_storage'


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

function borders.init()
   _G._crud.get_border_on_storage = get_border_on_storage
end

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',
        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 = 'read',
        replicasets = replicasets,
        timeout = opts.timeout,
    }
    local results, err, storages_info = call.map(vshard_router,
        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')	404✔
2	local errors = require('errors')	404✔
3
4	local const = require('crud.common.const')	404✔
5	local dev_checks = require('crud.common.dev_checks')	404✔
6	local call = require('crud.common.call')	404✔
7	local utils = require('crud.common.utils')	404✔
8	local schema = require('crud.common.schema')	404✔
9	local has_keydef, Keydef = pcall(require, 'crud.compare.keydef')	404✔
10	local select_comparators = require('crud.compare.comparators')	404✔
11
12	local BorderError = errors.new_class('BorderError', {capture_stack = false})	404✔
13
14	local borders = {}	404✔
15
16	local STAT_FUNC_NAME = '_crud.get_border_on_storage'	404✔
17
18
19	local function get_border_on_storage(border_name, space_name, index_id, field_names, fetch_latest_metadata)
20	dev_checks('string', 'string', 'number', '?table', '?boolean')	264✔
21
22	assert(border_name == 'min' or border_name == 'max')	264✔
23
24	local space = box.space[space_name]	264✔
25	if space == nil then	264✔
26	return nil, BorderError:new("Space %q doesn't exist", space_name)	×
27	end
28
29	local index = space.index[index_id]	264✔
30	if index == nil then	264✔
31	return nil, BorderError:new("Index %q of space doesn't exist", index_id, space_name)	×
32	end
33
34	local function get_index_border(index)
35	return index[border_name](index)	264✔
36	end
37
38	return schema.wrap_func_result(space, get_index_border, {index}, {	264✔
39	add_space_schema_hash = true,
40	field_names = field_names,	264✔
41	fetch_latest_metadata = fetch_latest_metadata,	264✔
42	})	264✔
43	end
44
45	function borders.init()	404✔
46	_G._crud.get_border_on_storage = get_border_on_storage	302✔
47	end
48
49	local is_closer
50
51	if has_keydef then	404✔
52	is_closer = function (compare_sign, keydef, tuple, res_tuple)
53	if res_tuple == nil then	108✔
54	return true	56✔
55	end
56
57	local cmp = keydef:compare(tuple, res_tuple)	104✔
58
59	return cmp * compare_sign > 0	52✔
60	end
61	else
62	is_closer = function (_, comparator, tuple, res_tuple)
63	if res_tuple == nil then	×
64	return true	×
65	end
66	return comparator(tuple, res_tuple)	×
67	end
68	end
69
70	local function call_get_border_on_router(vshard_router, border_name, space_name, index_name, opts)
71	checks('table', 'string', 'string', '?string\|number', {	312✔
72	timeout = '?number',
73	fields = '?table',
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)	312✔
79	if err ~= nil then	312✔
80	return nil, BorderError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
81	end
82	if space == nil then	312✔
83	return nil, BorderError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	72✔
84	end
85
86	local index
87	if index_name == nil then	276✔
88	index = space.index[0]	80✔
89	else
90	index = space.index[index_name]	196✔
91	end
92
93	if index == nil then	276✔
94	return nil,	144✔
95	BorderError:new("Index %q of space %q doesn't exist", index_name, space_name),	144✔
96	const.NEED_SCHEMA_RELOAD	144✔
97	end
98
99	local primary_index = space.index[0]	132✔
100
101	local cmp_key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)	132✔
102	local field_names = utils.enrich_field_names_with_cmp_key(opts.fields, cmp_key_parts, space:format())	264✔
103
104	local replicasets, err = vshard_router:routeall()	132✔
105	if err ~= nil then	132✔
106	return nil, BorderError:new("Failed to get router replicasets: %s", err)	×
107	end
108	local call_opts = {	132✔
109	mode = 'read',
110	replicasets = replicasets,	132✔
111	timeout = opts.timeout,	132✔
112	}
113	local results, err, storages_info = call.map(vshard_router,	264✔
114	STAT_FUNC_NAME,	132✔
115	{border_name, space_name, index.id, field_names, opts.fetch_latest_metadata},	132✔
116	call_opts
117	)	132✔
118
119	if err ~= nil then	132✔
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	132✔
124	local comparator
125	if has_keydef then	132✔
126	comparator = Keydef.new(space, field_names, index.id)	264✔
127	else
128	local tarantool_iter
129	if compare_sign > 0 then	×
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)	×
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	132✔
140	for _, storage_result in pairs(results) do	528✔
141	local storage_result = storage_result[1]	264✔
142	if storage_result.err ~= nil then	264✔
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)	×
146	if need_reload then	×
147	return nil, err_wrapped, const.NEED_SCHEMA_RELOAD	×
148	end
149
150	return nil, err_wrapped	×
151	end
152
153	local tuple = storage_result.res	264✔
154	if tuple ~= nil and is_closer(compare_sign, comparator, tuple, res_tuple) then	372✔
155	res_tuple = tuple	68✔
156	end
157	end
158
159	if opts.fetch_latest_metadata == true then	132✔
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,	8✔
164	storages_info, netbox_schema_version)	8✔
165	end
166
167	local result = utils.format_result({res_tuple}, space, field_names)	132✔
168
169	if opts.fields ~= nil then	132✔
170	result = utils.cut_rows(result.rows, result.metadata, opts.fields)	40✔
171	end
172
173	return result	132✔
174	end
175
176	local function get_border(border_name, space_name, index_name, opts)
177	opts = opts or {}	236✔
178	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	236✔
179	if err ~= nil then	236✔
180	return nil, BorderError:new(err)	32✔
181	end
182
183	return schema.wrap_func_reload(vshard_router, call_get_border_on_router,	220✔
184	border_name, space_name, index_name, opts	220✔
185	)	220✔
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)	404✔
213	return get_border('min', space_name, index_id, opts)	118✔
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)	404✔
241	return get_border('max', space_name, index_id, opts)	118✔
242	end
243
244	return borders	404✔

tarantool / crud / 5199459215

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