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 server hot reload techniques.
This entails an on_replace trigger duplication if hot reload occures.
Hot reload support will be done in separate task.

Closes #82

Author: 0x501D

ddl/cache.lua

@@ -0,0 +1,128 @@
+local fiber = require('fiber')
+
+local cache = nil
+
+local CACHE_LOCK_TIMEOUT  = 3
+local SPACE_NAME_IDX      = 1
+local SHARD_FUNC_NAME_IDX = 2
+local SHARD_FUNC_BODY_IDX = 3
+
+-- Function decorator that is used to prevent cache_build() from being
+-- called concurrently by different fibers.
+local function locked(f)
+    return function(...)
+        local ok = cache.lock:put(true, CACHE_LOCK_TIMEOUT)
+
+        if not ok then
+            error("cache lock timeout is exceeded")
+        end
+
+        local status, err = pcall(f, ...)
+        cache.lock:get()
+
+        if not status or err ~= nil then
+            return err
+        end
+    end
+end
+
+-- Build cache and setup "on_replace" trigger.
+--
+-- Cache structure format:
+-- cache = {
+--   spaces = {
+--     'space_name' = {
+--       raw = {},      -- raw sharding metadata, used for ddl.get()
+--       processed = {} -- table with parsed dot notation (like {'foo', 'bar'})
+--                      -- or a function ready to call (or a string with an error)
+--     }
+--   },
+--   lock -- locking based on fiber.channel()
+-- }
+local cache_build = locked(function()
+    -- clear cache
+    cache.spaces = {}
+
+    if box.space._ddl_sharding_func == nil then
+        return
+    end
+
+    for _, tuple in box.space._ddl_sharding_func:pairs() do
+        local space_name = tuple[SPACE_NAME_IDX]
+        local func_name = tuple[SHARD_FUNC_NAME_IDX]
+        local func_body = tuple[SHARD_FUNC_BODY_IDX]
+
+        cache.spaces[space_name] = {}
+        cache.spaces[space_name].sharding_func = { raw = tuple }
+
+        if func_body ~= nil then
+            local sharding_func, err = loadstring('return ' .. func_body)
+            if sharding_func == nil then
+                cache.spaces[space_name].sharding_func.processed =
+                    string.format("Body is incorrect in sharding_func for space (%s): %s",
+                    space_name, err)
+            else
+                cache.spaces[space_name].sharding_func.processed =
+                    sharding_func()
+            end
+        elseif func_name ~= nil then
+            local chunks = string.split(func_name, '.')
+            cache.spaces[space_name].sharding_func.processed = chunks
+        end
+    end
+
+    cache._schema_version = box.internal.schema_version()
+end)
+
+-- Rebuild cache if _ddl_sharding_func space changed.
+local function cache_set_trigger()
+    if box.space._ddl_sharding_func == nil then
+        return
+    end
+
+    local trigger_found = false
+
+    for _, func in pairs(box.space._ddl_sharding_func:on_replace()) do
+        if func == cache_build then
+            trigger_found = true
+            break
+        end
+    end
+
+    if not trigger_found then
+        box.space._ddl_sharding_func:on_replace(cache_build)
+    end
+end
+
+-- Get data from cache.
+-- Returns all cached data for "space_name".
+local function cache_get(space_name)
+    if space_name == nil then
+        return nil
+    end
+
+    local schema_version = box.internal.schema_version()
+
+    if not cache then
+        cache = {
+            spaces = {},
+            lock = fiber.channel(1)
+        }
+        cache_build()
+        cache_set_trigger()
+    end
+
+    -- rebuild cache if database schema changed
+    if schema_version ~= cache._schema_version then
+        cache_build()
+        cache_set_trigger()
+    end
+
+    return cache.spaces[space_name]
+end
+
+return {
+    internal = {
+        get = cache_get,
+    }
+}

ddl/get.lua

@@ -1,4 +1,5 @@
 local utils = require('ddl.utils')
+local cache = require('ddl.cache')
 local ddl_check = require('ddl.check')
 
 local function _get_index_field_path(space, index_part)
@@ -66,11 +67,25 @@ local function get_metadata(space_name, metadata_name)
 end
 
 local function get_sharding_func(space_name)
-    local record = get_metadata(space_name, "sharding_func")
-    if not record then
+    local record = cache.internal.get(space_name)
+
+    if not record or not record.sharding_func then
+        return nil
+    end
+
+    return record.sharding_func.processed
+end
+
+local function get_sharding_func_raw(space_name)
+    local record = cache.internal.get(space_name)
+
+    if not record or not record.sharding_func or not
+        record.sharding_func.raw then
         return nil
     end
 
+    record = record.sharding_func.raw
+
     if record.sharding_func_body ~= nil then
         return {body = record.sharding_func_body}
     end
@@ -97,7 +112,7 @@ local function get_space_schema(space_name)
     space_ddl.engine = box_space.engine
     space_ddl.format = box_space:format()
     space_ddl.sharding_key = get_sharding_key(space_name)
-    space_ddl.sharding_func = get_sharding_func(space_name)
+    space_ddl.sharding_func = get_sharding_func_raw(space_name)
     for _, field in ipairs(space_ddl.format) do
         if field.is_nullable == nil then
             field.is_nullable = false
@@ -115,21 +130,21 @@ local function get_space_schema(space_name)
 end
 
 local function prepare_sharding_func_for_call(space_name, sharding_func_def)
-    if type(sharding_func_def) == 'string' then
+    if type(sharding_func_def) == 'table' then
         local sharding_func = utils.get_G_function(sharding_func_def)
         if sharding_func ~= nil and
            ddl_check.internal.is_callable(sharding_func) == true then
             return sharding_func
         end
     end
 
-    if type(sharding_func_def) == 'table' then
-        local sharding_func, err = loadstring('return ' .. sharding_func_def.body)
-        if sharding_func == nil then
-            return nil, string.format(
-                "Body is incorrect in sharding_func for space (%s): %s", space_name, err)
-        end
-        return sharding_func()
+    if type(sharding_func_def) == 'function' then
+        return sharding_func_def
+    end
+
+    -- error from cache
+    if type(sharding_func_def) == 'string' then
+        return nil, sharding_func_def
     end
 
     return nil, string.format(

ddl/utils.lua

@@ -189,9 +189,19 @@ end
 -- split sharding func name in dot notation by dot
 -- foo.bar.baz -> chunks: foo bar baz
 -- foo -> chunks: foo
+--
+-- func_name parameter may be a string in dot notation or table
+-- if func_name type is of type table it is assumed that it is already split
 local function get_G_function(func_name)
-    local chunks = string.split(func_name, '.')
     local sharding_func = _G
+    local chunks
+
+    if type(func_name) == 'string' then
+        chunks = string.split(func_name, '.')
+    else
+        chunks = func_name
+    end
+
     -- check is the each chunk an identifier
     for _, chunk in pairs(chunks) do
         if not check_name_isident(chunk) or sharding_func == nil then