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

98.92

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

local insert = {}

local INSERT_FUNC_NAME = '_crud.insert_on_storage'

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

    -- 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
    return 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,
    })
end

function insert.init()
   _G._crud.insert_on_storage = insert_on_storage
end

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