Skip to content

Commit 42b0295

Browse files
0x501D0x501D
committed
Add caching of sharding function
The ddl.bucket_id() function needs to know a sharding function. It is costly to obtain the function declaration / definition stored in the _ddl_sharding_func space. This cache adds sharding function cache divided into two parts: raw and processed. Raw part is used for get_schema() method. Raw cache stored as is. Processed part is used for bucket_id(). Processed sharding_func cache entry may be: * table with parsed dot notation (like {'foo', 'bar'}) * function ready to call, this offloads using of loadstring() * string with an error Cache will be rebuilded if: * _ddl_sharding_func space changed: cache sets _ddl_sharding_func:on_replace trigger * schema changed: cache checks box.internal.schema_version changes This patch does not serve hot reload techniques. This entails an on_replace trigger duplication if hot reload occurs. Hot reload support will be done in separate task: #87 Closes #82
1 parent 4f0fbd1 commit 42b0295

File tree

4 files changed

+330
-11
lines changed

4 files changed

+330
-11
lines changed

‎ddl/cache.lua

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
local fiber = require('fiber')
2+
3+
local cache = nil
4+
5+
local CACHE_LOCK_TIMEOUT = 3
6+
local SPACE_NAME_IDX = 1
7+
local SHARD_FUNC_NAME_IDX = 2
8+
local SHARD_FUNC_BODY_IDX = 3
9+
10+
-- Function decorator that is used to prevent cache_build() from being
11+
-- called concurrently by different fibers.
12+
local function locked(f)
13+
return function(...)
14+
local ok = cache.lock:put(true, CACHE_LOCK_TIMEOUT)
15+
16+
if not ok then
17+
error("cache lock timeout is exceeded")
18+
end
19+
20+
local status, err = pcall(f, ...)
21+
cache.lock:get()
22+
23+
if not status or err ~= nil then
24+
return err
25+
end
26+
end
27+
end
28+
29+
-- Build cache.
30+
--
31+
-- Cache structure format:
32+
-- cache = {
33+
-- spaces = {
34+
-- 'space_name' = {
35+
-- raw = {}, -- raw sharding metadata, used for ddl.get()
36+
-- processed = {} -- table with parsed dot notation (like {'foo', 'bar'})
37+
-- -- or a function ready to call (or a string with an error)
38+
-- }
39+
-- },
40+
-- lock -- locking based on fiber.channel()
41+
-- schema_version -- current schema version
42+
-- }
43+
--
44+
-- function returns nothing
45+
local cache_build = locked(function()
46+
-- clear cache
47+
cache.spaces = {}
48+
49+
if box.space._ddl_sharding_func == nil then
50+
return
51+
end
52+
53+
for _, tuple in box.space._ddl_sharding_func:pairs() do
54+
local space_name = tuple[SPACE_NAME_IDX]
55+
local func_name = tuple[SHARD_FUNC_NAME_IDX]
56+
local func_body = tuple[SHARD_FUNC_BODY_IDX]
57+
58+
cache.spaces[space_name] = {
59+
raw = tuple
60+
}
61+
62+
if func_body ~= nil then
63+
local sharding_func, err = loadstring('return ' .. func_body)
64+
if sharding_func == nil then
65+
cache.spaces[space_name].processed =
66+
string.format("Body is incorrect in sharding_func for space (%s): %s",
67+
space_name, err)
68+
else
69+
cache.spaces[space_name].processed =
70+
sharding_func()
71+
end
72+
elseif func_name ~= nil then
73+
local chunks = string.split(func_name, '.')
74+
cache.spaces[space_name].processed = chunks
75+
end
76+
end
77+
78+
end)
79+
80+
-- Rebuild cache if _ddl_sharding_func space changed.
81+
local function cache_set_trigger()
82+
if box.space._ddl_sharding_func == nil then
83+
return
84+
end
85+
86+
local trigger_found = false
87+
88+
for _, func in pairs(box.space._ddl_sharding_func:on_replace()) do
89+
if func == cache_build then
90+
trigger_found = true
91+
break
92+
end
93+
end
94+
95+
if not trigger_found then
96+
box.space._ddl_sharding_func:on_replace(cache_build)
97+
end
98+
end
99+
100+
-- Get data from cache.
101+
-- Returns all cached data for "space_name" or nil.
102+
local function cache_get(space_name)
103+
if space_name == nil then
104+
return nil
105+
end
106+
107+
-- using tarantool internal API.
108+
-- this is not reliable, but it is the only way to track
109+
-- schema_version changes. Fix it if a public method appears:
110+
-- https://github.com/tarantool/tarantool/issues/6544
111+
local schema_version = box.internal.schema_version()
112+
113+
if not cache then
114+
cache = {
115+
lock = fiber.channel(1)
116+
}
117+
cache_build()
118+
cache_set_trigger()
119+
cache.schema_version = schema_version
120+
end
121+
122+
-- rebuild cache if database schema changed
123+
if schema_version ~= cache.schema_version then
124+
cache_build()
125+
cache_set_trigger()
126+
cache.schema_version = schema_version
127+
end
128+
129+
return cache.spaces[space_name]
130+
end
131+
132+
return {
133+
internal = {
134+
get = cache_get,
135+
}
136+
}

‎ddl/get.lua

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
local utils = require('ddl.utils')
2+
local cache = require('ddl.cache')
23
local ddl_check = require('ddl.check')
34

45
local function _get_index_field_path(space, index_part)
@@ -66,11 +67,24 @@ local function get_metadata(space_name, metadata_name)
6667
end
6768

6869
local function get_sharding_func(space_name)
69-
local record = get_metadata(space_name, "sharding_func")
70+
local record = cache.internal.get(space_name)
71+
7072
if not record then
7173
return nil
7274
end
7375

76+
return record.processed
77+
end
78+
79+
local function get_sharding_func_raw(space_name)
80+
local record = cache.internal.get(space_name)
81+
82+
if not record or not record.raw then
83+
return nil
84+
end
85+
86+
record = record.raw
87+
7488
if record.sharding_func_body ~= nil then
7589
return {body = record.sharding_func_body}
7690
end
@@ -97,7 +111,7 @@ local function get_space_schema(space_name)
97111
space_ddl.engine = box_space.engine
98112
space_ddl.format = box_space:format()
99113
space_ddl.sharding_key = get_sharding_key(space_name)
100-
space_ddl.sharding_func = get_sharding_func(space_name)
114+
space_ddl.sharding_func = get_sharding_func_raw(space_name)
101115
for _, field in ipairs(space_ddl.format) do
102116
if field.is_nullable == nil then
103117
field.is_nullable = false
@@ -115,21 +129,21 @@ local function get_space_schema(space_name)
115129
end
116130

117131
local function prepare_sharding_func_for_call(space_name, sharding_func_def)
118-
if type(sharding_func_def) == 'string' then
132+
if type(sharding_func_def) == 'table' then
119133
local sharding_func = utils.get_G_function(sharding_func_def)
120134
if sharding_func ~= nil and
121135
ddl_check.internal.is_callable(sharding_func) == true then
122136
return sharding_func
123137
end
124138
end
125139

126-
if type(sharding_func_def) == 'table' then
127-
local sharding_func, err = loadstring('return ' .. sharding_func_def.body)
128-
if sharding_func == nil then
129-
return nil, string.format(
130-
"Body is incorrect in sharding_func for space (%s): %s", space_name, err)
131-
end
132-
return sharding_func()
140+
if type(sharding_func_def) == 'function' then
141+
return sharding_func_def
142+
end
143+
144+
-- error from cache
145+
if type(sharding_func_def) == 'string' then
146+
return nil, sharding_func_def
133147
end
134148

135149
return nil, string.format(

‎ddl/utils.lua

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,9 +189,19 @@ end
189189
-- split sharding func name in dot notation by dot
190190
-- foo.bar.baz -> chunks: foo bar baz
191191
-- foo -> chunks: foo
192+
--
193+
-- func_name parameter may be a string in dot notation or table
194+
-- if func_name type is of type table it is assumed that it is already split
192195
local function get_G_function(func_name)
193-
local chunks = string.split(func_name, '.')
194196
local sharding_func = _G
197+
local chunks
198+
199+
if type(func_name) == 'string' then
200+
chunks = string.split(func_name, '.')
201+
else
202+
chunks = func_name
203+
end
204+
195205
-- check is the each chunk an identifier
196206
for _, chunk in pairs(chunks) do
197207
if not check_name_isident(chunk) or sharding_func == nil then

‎test/cache_test.lua

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
#!/usr/bin/env tarantool
2+
3+
local t = require('luatest')
4+
local db = require('test.db')
5+
local ddl = require('ddl')
6+
local cache = require('ddl.cache')
7+
local helper = require('test.helper')
8+
9+
local SPACE_NAME_IDX = 1
10+
local SHARD_FUNC_NAME_IDX = 2
11+
local SHARD_FUNC_BODY_IDX = 3
12+
13+
local primary_index = {
14+
type = 'HASH',
15+
unique = true,
16+
parts = {
17+
{path = 'string_nonnull', is_nullable = false, type = 'string'},
18+
{path = 'unsigned_nonnull', is_nullable = false, type = 'unsigned'},
19+
},
20+
name = 'primary'
21+
}
22+
23+
local bucket_id_idx = {
24+
type = 'TREE',
25+
unique = false,
26+
parts = {{path = 'bucket_id', type = 'unsigned', is_nullable = false}},
27+
name = 'bucket_id'
28+
}
29+
30+
local func_body_first = 'function() return 42 end'
31+
local func_body_second = 'function() return 24 end'
32+
33+
local function rebuild_db(g)
34+
db.drop_all()
35+
36+
g.space = {
37+
engine = 'memtx',
38+
is_local = true,
39+
temporary = false,
40+
format = table.deepcopy(helper.test_space_format())
41+
}
42+
table.insert(g.space.format, 1, {
43+
name = 'bucket_id', type = 'unsigned', is_nullable = false
44+
})
45+
46+
g.space.indexes = {
47+
table.deepcopy(primary_index),
48+
table.deepcopy(bucket_id_idx)
49+
}
50+
g.space.sharding_key = {'unsigned_nonnull', 'integer_nonnull'}
51+
g.schema = {
52+
spaces = {
53+
space = g.space,
54+
}
55+
}
56+
end
57+
58+
local g = t.group()
59+
g.before_all(db.init)
60+
g.before_each(function()
61+
rebuild_db(g)
62+
end)
63+
64+
function g.test_cache_processed_func_body()
65+
g.schema.spaces.space.sharding_func = {
66+
body = func_body_first
67+
}
68+
local ok, err = ddl.set_schema(g.schema)
69+
t.assert_equals(err, nil)
70+
t.assert_equals(ok, true)
71+
72+
local res = cache.internal.get('space')
73+
t.assert(res)
74+
t.assert(res.processed)
75+
res = res.processed
76+
t.assert(type(res) == 'function')
77+
t.assert_equals(res(), 42)
78+
end
79+
80+
function g.test_cache_processed_func_name()
81+
local sharding_func_name = 'sharding_func'
82+
rawset(_G, sharding_func_name, function(key) return key end)
83+
g.schema.spaces.space.sharding_func = sharding_func_name
84+
85+
local ok, err = ddl.set_schema(g.schema)
86+
t.assert_equals(err, nil)
87+
t.assert_equals(ok, true)
88+
89+
local res = cache.internal.get('space')
90+
t.assert(res)
91+
t.assert(res.processed)
92+
res = res.processed
93+
t.assert(type(res) == 'table')
94+
t.assert_equals(res[1], 'sharding_func')
95+
96+
rawset(_G, sharding_func_name, nil)
97+
end
98+
99+
function g.test_cache_schema_changed()
100+
g.schema.spaces.space.sharding_func = {
101+
body = func_body_first
102+
}
103+
local ok, err = ddl.set_schema(g.schema)
104+
t.assert_equals(err, nil)
105+
t.assert_equals(ok, true)
106+
107+
local res = cache.internal.get('space')
108+
t.assert(res)
109+
t.assert(res.raw)
110+
res = res.raw
111+
t.assert_equals(res[SPACE_NAME_IDX], 'space')
112+
t.assert_equals(res[SHARD_FUNC_NAME_IDX], nil)
113+
t.assert_equals(res[SHARD_FUNC_BODY_IDX], func_body_first)
114+
115+
rebuild_db(g)
116+
117+
g.schema.spaces.space.sharding_func = {
118+
body = func_body_second
119+
}
120+
local ok, err = ddl.set_schema(g.schema)
121+
t.assert_equals(err, nil)
122+
t.assert_equals(ok, true)
123+
124+
local res = cache.internal.get('space')
125+
t.assert(res)
126+
t.assert(res.raw)
127+
res = res.raw
128+
t.assert_equals(res[SPACE_NAME_IDX], 'space')
129+
t.assert_equals(res[SHARD_FUNC_NAME_IDX], nil)
130+
t.assert_equals(res[SHARD_FUNC_BODY_IDX], func_body_second)
131+
end
132+
133+
function g.test_cache_space_updated()
134+
g.schema.spaces.space.sharding_func = {
135+
body = func_body_first
136+
}
137+
local ok, err = ddl.set_schema(g.schema)
138+
t.assert_equals(err, nil)
139+
t.assert_equals(ok, true)
140+
141+
local res = cache.internal.get('space')
142+
t.assert(res)
143+
t.assert(res.raw)
144+
res = res.raw
145+
t.assert_equals(res[SPACE_NAME_IDX], 'space')
146+
t.assert_equals(res[SHARD_FUNC_NAME_IDX], nil)
147+
t.assert_equals(res[SHARD_FUNC_BODY_IDX], func_body_first)
148+
149+
box.space._ddl_sharding_func
150+
:update({'space'}, {{'=', SHARD_FUNC_BODY_IDX, func_body_second}})
151+
152+
local res = cache.internal.get('space')
153+
t.assert(res)
154+
t.assert(res.raw)
155+
res = res.raw
156+
t.assert_equals(res[SPACE_NAME_IDX], 'space')
157+
t.assert_equals(res[SHARD_FUNC_NAME_IDX], nil)
158+
t.assert_equals(res[SHARD_FUNC_BODY_IDX], func_body_second)
159+
end

0 commit comments

Comments
 (0)