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

90.65

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

local update = {}

local UPDATE_FUNC_NAME = '_crud.update_on_storage'

local function update_on_storage(space_name, key, operations, field_names, opts)
    dev_checks('string', '?', 'table', '?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, 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

    -- 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 then
        return nil, err
    end

    if res.err == nil then
        return res, nil
    end

    -- 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
    if utils.is_field_not_found(res.err.code) then
        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,
        })
    end

    return res, err
end

function update.init()
   _G._crud.update_on_storage = update_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_update_on_router(vshard_router, space_name, key, user_operations, opts)
    dev_checks('table', 'string', '?', 'table', {
        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, 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

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