7320500616

Committed 25 Dec 2023 08:41AM UTC coverage: 89.158% (+0.01%) from 89.145%

Build # 7320500616

Build Type

Pull #404

github

Committed by

DifferentialOrange

Commit Message

update: propagate new options if fields not exist

If Tarantool older than 2.8.1 [1] was used to update nullable fields,
the workaround was executed to perform this operation. If `noreturn` or
`fetch_latest_metadata` options were set, they were ignored on storage
side before this patch.

1. https://github.com/tarantool/tarantool/issues/3378

Pull Request Pull Request #404: Support vshard names as keys

Run Details

25 of 30 new or added lines in 5 files covered. (83.33%)

1 existing line in 1 file now uncovered.

4679 of 5248 relevant lines covered (89.16%)

5996.97 hits per line

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

14.75

/crud/readview.lua

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

local const = require('crud.common.const')
local stash = require('crud.common.stash')
local utils = require('crud.common.utils')
local sharding = require('crud.common.sharding')
local select_executor = require('crud.select.executor')
local select_filters = require('crud.compare.filters')
local dev_checks = require('crud.common.dev_checks')
local schema = require('crud.common.schema')
local stats = require('crud.stats')

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

local has_merger = (utils.tarantool_supports_external_merger() and
    package.search('tuple.merger')) or utils.tarantool_has_builtin_merger()

local OPEN_FUNC_NAME = 'readview_open_on_storage'
local CRUD_OPEN_FUNC_NAME = utils.get_storage_call(OPEN_FUNC_NAME)
local SELECT_FUNC_NAME = 'select_readview_on_storage'
local CLOSE_FUNC_NAME = 'readview_close_on_storage'
local CRUD_CLOSE_FUNC_NAME = utils.get_storage_call(CLOSE_FUNC_NAME)

if (not utils.tarantool_version_at_least(2, 11, 0))
or (tarantool.package ~= 'Tarantool Enterprise') or (not has_merger) then
    return {
        new = function() return nil,
        ReadviewError:new("Tarantool does not support readview") end,
        init = function() return nil end}
end
local select = require('crud.select.compat.select')

local readview = {}


local function readview_open_on_storage(readview_name)
    if not utils.tarantool_version_at_least(2, 11, 0) or
    tarantool.package ~= 'Tarantool Enterprise' then
        ReadviewError:assert(false, ("Tarantool does not support readview"))
    end
    -- We store readview in stash because otherwise gc will delete it.
    -- e.g master switch.
    local read_view = box.read_view.open({name = readview_name})
    local stash_readview = stash.get(stash.name.storage_readview)
    stash_readview[read_view.id] = read_view

    if read_view == nil then
        ReadviewError:assert(false, ("Error creating readview"))
    end

    local replica_info = {}
    replica_info.uuid = box.info().uuid
    replica_info.id = read_view.id

    return replica_info, nil
end

local function readview_close_on_storage(readview_uuid)
    dev_checks('table')

    local list = box.read_view.list()
    local readview_id
    for _, replica_info in pairs(readview_uuid) do
        if replica_info.uuid == box.info().uuid then
            readview_id = replica_info.id
        end
    end

    for k,v in pairs(list) do
        if v.id == readview_id then
            list[k]:close()
            local stash_readview = stash.get(stash.name.storage_readview)
            stash_readview[readview_id] = nil
            return true
        end
    end

    return false
end

local function select_readview_on_storage(space_name, index_id, conditions, opts)
    dev_checks('string', 'number', '?table', {
        scan_value = 'table',
        after_tuple = '?table',
        tarantool_iter = 'number',
        limit = 'number',
        scan_condition_num = '?number',
        field_names = '?table',
        sharding_key_hash = '?number',
        sharding_func_hash = '?number',
        skip_sharding_hash_check = '?boolean',
        yield_every = '?number',
        fetch_latest_metadata = '?boolean',
        readview_id = 'number',
    })

    local cursor = {}
    if opts.fetch_latest_metadata then
        local replica_schema_version
        if box.info.schema_version ~= nil then
            replica_schema_version = box.info.schema_version
        else
            replica_schema_version = box.internal.schema_version()
        end
        cursor.storage_info = {
            replica_uuid = box.info().uuid,
            replica_schema_version = replica_schema_version,
        }
    end

    local list = box.read_view.list()
    local space_readview

    for k,v in pairs(list) do
        if v.id == opts.readview_id then
            space_readview = list[k].space[space_name]
        end
    end

    if space_readview == nil then
        return cursor, ReadviewError:new("Space %q doesn't exist", space_name)
    end

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

    local index_readview = space_readview.index[index_id]
    if index_readview == nil then
        return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)
    end
    local index = space.index[index_id]
    if index == nil then
        return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)
    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

    local filter_func, err = select_filters.gen_func(space, conditions, {
        tarantool_iter = opts.tarantool_iter,
        scan_condition_num = opts.scan_condition_num,
    })
    if err ~= nil then
        return cursor, ReadviewError:new("Failed to generate tuples filter: %s", err)
    end

    -- execute select
    local resp, err = select_executor.execute(space, index, filter_func, {
        scan_value = opts.scan_value,
        after_tuple = opts.after_tuple,
        tarantool_iter = opts.tarantool_iter,
        limit = opts.limit,
        yield_every = opts.yield_every,
        readview = true,
        readview_index = index_readview,
    })
    if err ~= nil then
        return cursor, ReadviewError:new("Failed to execute select: %s", err)
    end

    if resp.tuples_fetched < opts.limit or opts.limit == 0 then
        cursor.is_end = true
    else
        local last_tuple = resp.tuples[#resp.tuples]
        cursor.after_tuple = last_tuple:totable()
    end

    cursor.stats = {
        tuples_lookup = resp.tuples_lookup,
        tuples_fetched = resp.tuples_fetched,
    }

    -- getting tuples with user defined fields (if `fields` option is specified)
    -- and fields that are needed for comparison on router (primary key + scan key)
    local filtered_tuples = schema.filter_tuples_fields(resp.tuples, opts.field_names)

    local result = {cursor, filtered_tuples}

    local select_module_compat_info = stash.get(stash.name.select_module_compat_info)
    if not select_module_compat_info.has_merger then
        if opts.fetch_latest_metadata then
            result[3] = cursor.storage_info.replica_schema_version
        end
    end

    return unpack(result)
end

local Readview_obj = {}
Readview_obj.__index = Readview_obj

local select_call = stats.wrap(select.call, stats.op.SELECT)

function Readview_obj:select(space_name, user_conditions, opts)
    opts = opts or {}
    opts.readview = true
    opts.readview_uuid = self._uuid

    if self.opened == false then
        return nil, ReadviewError:new("Read view is closed")
    end

    return select_call(space_name, user_conditions, opts)
end

local pairs_call = stats.wrap(select.pairs, stats.op.SELECT, {pairs = true})

function Readview_obj:pairs(space_name, user_conditions, opts)
    opts = opts or {}
    opts.readview = true
    opts.readview_uuid = self._uuid

    if self.opened == false then
        return nil, ReadviewError:new("Read view is closed")
    end

    return pairs_call(space_name, user_conditions, opts)
end

function readview.init(user)
    utils.init_storage_call(user, OPEN_FUNC_NAME, readview_open_on_storage)
    utils.init_storage_call(user, CLOSE_FUNC_NAME, readview_close_on_storage)
    utils.init_storage_call(user, SELECT_FUNC_NAME, select_readview_on_storage)
 end

function Readview_obj:close(opts)
    checks('table', {
        timeout = '?number',
    })
    opts = opts or {}
    if self.opened == false then
        return
    end

    local vshard_router, err = utils.get_vshard_router_instance(nil)
    if err ~= nil then
        return ReadviewError:new(err)
    end

    local replicasets, err = vshard_router:routeall()
    if err ~= nil then
        return ReadviewError:new(err)
    end

    if opts.timeout == nil then
        opts.timeout = const.DEFAULT_VSHARD_CALL_TIMEOUT
    end

    local errors = {}
    for _, replicaset in pairs(replicasets) do
        for _, replica in pairs(replicaset.replicas) do
            for _, value in pairs(self._uuid) do
                if replica.uuid == value.uuid then
                    local replica_result, replica_err = replica.conn:call(CRUD_CLOSE_FUNC_NAME,
                    {self._uuid}, {timeout = opts.timeout})
                    if replica_err ~= nil then
                        table.insert(errors, ReadviewError:new("Failed to close Readview on storage: %s", replica_err))
                    end
                    if replica_err == nil and (not replica_result) then
                        table.insert(errors, ReadviewError:new("Readview was not found on storage: %s", replica.uuid))
                    end
                end
            end
        end
    end

    if next(errors) ~= nil then
        return errors
    end

    self.opened = false
    return nil

end

function Readview_obj:__gc()
    fiber.new(self.close, self)
end

function Readview_obj.create(vshard_router, opts)
    local readview = {}

    -- For tarantool lua (and lua 5.1) __gc metamethod only works for cdata types.
    -- So in order to create a proper GC hook, we need to create cdata with
    -- __gc call.
    -- __gc call for this cdata will be a __gc call for our readview.
    -- https://github.com/tarantool/tarantool/issues/5770
    local proxy = newproxy(true)
    getmetatable(proxy).__gc = function(_) Readview_obj.__gc(readview) end
    readview[proxy] = true
    setmetatable(readview, Readview_obj)

    readview._name = opts.name
    local results, err, err_uuid = vshard_router:map_callrw(CRUD_OPEN_FUNC_NAME,
        {readview._name}, {timeout = opts.timeout})
    if err ~= nil then
        return nil,
        ReadviewError:new("Failed to call readview_open_on_storage on storage-side: storage uuid: %s err: %s",
        err_uuid, err)
    end

    local uuid = {}
    local errors = {}
    for _, replicaset_results in pairs(results) do
        for _, replica_result in pairs(replicaset_results) do
            table.insert(uuid, replica_result)
        end
    end

    readview._uuid = uuid
    readview.opened = true

    if next(errors) ~= nil then
        return nil, errors
    end
    return readview, nil
 end

function readview.new(opts)
    checks({
        name = '?string',
        timeout = '?number',
    })
    opts = opts or {}
    local vshard_router, err = utils.get_vshard_router_instance(nil)
    if err ~= nil then
        return nil, ReadviewError:new(err)
    end

    return Readview_obj.create(vshard_router, opts)
end


return readview

1	local fiber = require('fiber')	360✔
2	local checks = require('checks')	360✔
3	local errors = require('errors')	360✔
4	local tarantool = require('tarantool')	360✔
5
6	local const = require('crud.common.const')	360✔
7	local stash = require('crud.common.stash')	360✔
8	local utils = require('crud.common.utils')	360✔
9	local sharding = require('crud.common.sharding')	360✔
10	local select_executor = require('crud.select.executor')	360✔
11	local select_filters = require('crud.compare.filters')	360✔
12	local dev_checks = require('crud.common.dev_checks')	360✔
13	local schema = require('crud.common.schema')	360✔
14	local stats = require('crud.stats')	360✔
15
16	local ReadviewError = errors.new_class('ReadviewError', {capture_stack = false})	360✔
17
18	local has_merger = (utils.tarantool_supports_external_merger() and	720✔
19	package.search('tuple.merger')) or utils.tarantool_has_builtin_merger()	720✔
20
21	local OPEN_FUNC_NAME = 'readview_open_on_storage'	360✔
22	local CRUD_OPEN_FUNC_NAME = utils.get_storage_call(OPEN_FUNC_NAME)	360✔
23	local SELECT_FUNC_NAME = 'select_readview_on_storage'	360✔
24	local CLOSE_FUNC_NAME = 'readview_close_on_storage'	360✔
25	local CRUD_CLOSE_FUNC_NAME = utils.get_storage_call(CLOSE_FUNC_NAME)	360✔
26
27	if (not utils.tarantool_version_at_least(2, 11, 0))	720✔
28	or (tarantool.package ~= 'Tarantool Enterprise') or (not has_merger) then	360✔
29	return {	360✔
30	new = function() return nil,	1✔
31	ReadviewError:new("Tarantool does not support readview") end,	362✔
32	init = function() return nil end}	643✔
33	end
34	local select = require('crud.select.compat.select')	×
35
36	local readview = {}	×
37
38
39	local function readview_open_on_storage(readview_name)
40	if not utils.tarantool_version_at_least(2, 11, 0) or	×
41	tarantool.package ~= 'Tarantool Enterprise' then	×
42	ReadviewError:assert(false, ("Tarantool does not support readview"))	×
43	end
44	-- We store readview in stash because otherwise gc will delete it.
45	-- e.g master switch.
46	local read_view = box.read_view.open({name = readview_name})	×
47	local stash_readview = stash.get(stash.name.storage_readview)	×
48	stash_readview[read_view.id] = read_view	×
49
50	if read_view == nil then	×
51	ReadviewError:assert(false, ("Error creating readview"))	×
52	end
53
54	local replica_info = {}	×
55	replica_info.uuid = box.info().uuid	×
56	replica_info.id = read_view.id	×
57
58	return replica_info, nil	×
59	end
60
61	local function readview_close_on_storage(readview_uuid)
62	dev_checks('table')	×
63
64	local list = box.read_view.list()	×
65	local readview_id
66	for _, replica_info in pairs(readview_uuid) do	×
67	if replica_info.uuid == box.info().uuid then	×
68	readview_id = replica_info.id	×
69	end
70	end
71
72	for k,v in pairs(list) do	×
73	if v.id == readview_id then	×
74	list[k]:close()	×
75	local stash_readview = stash.get(stash.name.storage_readview)	×
76	stash_readview[readview_id] = nil	×
77	return true	×
78	end
79	end
80
81	return false	×
82	end
83
84	local function select_readview_on_storage(space_name, index_id, conditions, opts)
85	dev_checks('string', 'number', '?table', {	×
86	scan_value = 'table',
87	after_tuple = '?table',
88	tarantool_iter = 'number',
89	limit = 'number',
90	scan_condition_num = '?number',
91	field_names = '?table',
92	sharding_key_hash = '?number',
93	sharding_func_hash = '?number',
94	skip_sharding_hash_check = '?boolean',
95	yield_every = '?number',
96	fetch_latest_metadata = '?boolean',
97	readview_id = 'number',
98	})
99
100	local cursor = {}	×
101	if opts.fetch_latest_metadata then	×
102	local replica_schema_version
103	if box.info.schema_version ~= nil then	×
104	replica_schema_version = box.info.schema_version	×
105	else
106	replica_schema_version = box.internal.schema_version()	×
107	end
108	cursor.storage_info = {	×
109	replica_uuid = box.info().uuid,
110	replica_schema_version = replica_schema_version,
111	}
112	end
113
114	local list = box.read_view.list()	×
115	local space_readview
116
117	for k,v in pairs(list) do	×
118	if v.id == opts.readview_id then	×
119	space_readview = list[k].space[space_name]	×
120	end
121	end
122
123	if space_readview == nil then	×
124	return cursor, ReadviewError:new("Space %q doesn't exist", space_name)	×
125	end
126
127	local space = box.space[space_name]	×
128	if space == nil then	×
129	return cursor, ReadviewError:new("Space %q doesn't exist", space_name)	×
130	end
131	space_readview.format = space:format()	×
132
133	local index_readview = space_readview.index[index_id]	×
134	if index_readview == nil then	×
135	return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)	×
136	end
137	local index = space.index[index_id]	×
138	if index == nil then	×
139	return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)	×
140	end
141
142	local _, err = sharding.check_sharding_hash(space_name,	×
143	opts.sharding_func_hash,	×
144	opts.sharding_key_hash,	×
145	opts.skip_sharding_hash_check)	×
146
147	if err ~= nil then	×
148	return nil, err	×
149	end
150
151	local filter_func, err = select_filters.gen_func(space, conditions, {	×
152	tarantool_iter = opts.tarantool_iter,
153	scan_condition_num = opts.scan_condition_num,
154	})
155	if err ~= nil then	×
156	return cursor, ReadviewError:new("Failed to generate tuples filter: %s", err)	×
157	end
158
159	-- execute select
160	local resp, err = select_executor.execute(space, index, filter_func, {	×
161	scan_value = opts.scan_value,
162	after_tuple = opts.after_tuple,
163	tarantool_iter = opts.tarantool_iter,
164	limit = opts.limit,
165	yield_every = opts.yield_every,
166	readview = true,
167	readview_index = index_readview,
168	})
169	if err ~= nil then	×
170	return cursor, ReadviewError:new("Failed to execute select: %s", err)	×
171	end
172
173	if resp.tuples_fetched < opts.limit or opts.limit == 0 then	×
174	cursor.is_end = true	×
175	else
176	local last_tuple = resp.tuples[#resp.tuples]	×
177	cursor.after_tuple = last_tuple:totable()	×
178	end
179
180	cursor.stats = {	×
181	tuples_lookup = resp.tuples_lookup,
182	tuples_fetched = resp.tuples_fetched,
183	}
184
185	-- getting tuples with user defined fields (if `fields` option is specified)
186	-- and fields that are needed for comparison on router (primary key + scan key)
187	local filtered_tuples = schema.filter_tuples_fields(resp.tuples, opts.field_names)	×
188
189	local result = {cursor, filtered_tuples}	×
190
191	local select_module_compat_info = stash.get(stash.name.select_module_compat_info)	×
192	if not select_module_compat_info.has_merger then	×
193	if opts.fetch_latest_metadata then	×
194	result[3] = cursor.storage_info.replica_schema_version	×
195	end
196	end
197
198	return unpack(result)	×
199	end
200
201	local Readview_obj = {}	×
202	Readview_obj.__index = Readview_obj	×
203
204	local select_call = stats.wrap(select.call, stats.op.SELECT)	×
205
206	function Readview_obj:select(space_name, user_conditions, opts)	×
207	opts = opts or {}	×
208	opts.readview = true	×
209	opts.readview_uuid = self._uuid	×
210
211	if self.opened == false then	×
212	return nil, ReadviewError:new("Read view is closed")	×
213	end
214
215	return select_call(space_name, user_conditions, opts)	×
216	end
217
218	local pairs_call = stats.wrap(select.pairs, stats.op.SELECT, {pairs = true})	×
219
220	function Readview_obj:pairs(space_name, user_conditions, opts)	×
221	opts = opts or {}	×
222	opts.readview = true	×
223	opts.readview_uuid = self._uuid	×
224
225	if self.opened == false then	×
226	return nil, ReadviewError:new("Read view is closed")	×
227	end
228
229	return pairs_call(space_name, user_conditions, opts)	×
230	end
231
232	function readview.init(user)	×
233	utils.init_storage_call(user, OPEN_FUNC_NAME, readview_open_on_storage)	×
234	utils.init_storage_call(user, CLOSE_FUNC_NAME, readview_close_on_storage)	×
235	utils.init_storage_call(user, SELECT_FUNC_NAME, select_readview_on_storage)	×
236	end
237
238	function Readview_obj:close(opts)	×
239	checks('table', {	×
240	timeout = '?number',
241	})
242	opts = opts or {}	×
243	if self.opened == false then	×
244	return	×
245	end
246
247	local vshard_router, err = utils.get_vshard_router_instance(nil)	×
248	if err ~= nil then	×
249	return ReadviewError:new(err)	×
250	end
251
252	local replicasets, err = vshard_router:routeall()	×
253	if err ~= nil then	×
254	return ReadviewError:new(err)	×
255	end
256
257	if opts.timeout == nil then	×
258	opts.timeout = const.DEFAULT_VSHARD_CALL_TIMEOUT	×
259	end
260
261	local errors = {}	×
262	for _, replicaset in pairs(replicasets) do	×
NEW 263	for _, replica in pairs(replicaset.replicas) do	×
264	for _, value in pairs(self._uuid) do	×
NEW 265	if replica.uuid == value.uuid then	×
266	local replica_result, replica_err = replica.conn:call(CRUD_CLOSE_FUNC_NAME,	×
267	{self._uuid}, {timeout = opts.timeout})	×
268	if replica_err ~= nil then	×
269	table.insert(errors, ReadviewError:new("Failed to close Readview on storage: %s", replica_err))	×
270	end
271	if replica_err == nil and (not replica_result) then	×
NEW 272	table.insert(errors, ReadviewError:new("Readview was not found on storage: %s", replica.uuid))	×
273	end
274	end
275	end
276	end
277	end
278
279	if next(errors) ~= nil then	×
280	return errors	×
281	end
282
283	self.opened = false	×
284	return nil	×
285
286	end
287
288	function Readview_obj:__gc()	×
289	fiber.new(self.close, self)	×
290	end
291
292	function Readview_obj.create(vshard_router, opts)	×
293	local readview = {}	×
294
295	-- For tarantool lua (and lua 5.1) __gc metamethod only works for cdata types.
296	-- So in order to create a proper GC hook, we need to create cdata with
297	-- __gc call.
298	-- __gc call for this cdata will be a __gc call for our readview.
299	-- https://github.com/tarantool/tarantool/issues/5770
300	local proxy = newproxy(true)	×
301	getmetatable(proxy).__gc = function(_) Readview_obj.__gc(readview) end	×
302	readview[proxy] = true	×
303	setmetatable(readview, Readview_obj)	×
304
305	readview._name = opts.name	×
306	local results, err, err_uuid = vshard_router:map_callrw(CRUD_OPEN_FUNC_NAME,	×
307	{readview._name}, {timeout = opts.timeout})	×
308	if err ~= nil then	×
309	return nil,	×
310	ReadviewError:new("Failed to call readview_open_on_storage on storage-side: storage uuid: %s err: %s",	×
311	err_uuid, err)	×
312	end
313
314	local uuid = {}	×
315	local errors = {}	×
316	for _, replicaset_results in pairs(results) do	×
317	for _, replica_result in pairs(replicaset_results) do	×
318	table.insert(uuid, replica_result)	×
319	end
320	end
321
322	readview._uuid = uuid	×
323	readview.opened = true	×
324
325	if next(errors) ~= nil then	×
326	return nil, errors	×
327	end
328	return readview, nil	×
329	end
330
331	function readview.new(opts)	×
332	checks({	×
333	name = '?string',
334	timeout = '?number',
335	})
336	opts = opts or {}	×
337	local vshard_router, err = utils.get_vshard_router_instance(nil)	×
338	if err ~= nil then	×
339	return nil, ReadviewError:new(err)	×
340	end
341
342	return Readview_obj.create(vshard_router, opts)	×
343	end
344
345
346	return readview	×

tarantool / crud / 7320500616

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