Some new features

Iterations over Secondary Indexes, ascending, start_element and
stop_iteration function

Author: ArtDu

expirationd.lua

@@ -45,7 +45,9 @@ local constants = {
     -- factor for recalculation of vinyl space size
     default_vinyl_assumed_space_len_factor = 2,
     -- default function on full scan
-    default_on_full_scan = function(...) end
+    default_on_full_scan = function(...) end,
+    -- default iterating over the loop will go in ascending index
+    iterator = "GE"
 }
 
 -- ========================================================================= --
@@ -57,10 +59,10 @@ local constants = {
 -- ------------------------------------------------------------------------- --
 
 -- get all fields in primary key(composite possible) from tuple
-local function construct_key(space_id, tuple)
+local function construct_key(expire_index, tuple)
     return fun.map(
         function(x) return tuple[x.fieldno] end,
-        box.space[space_id].index[0].parts
+        expire_index.parts
     ):totable()
 end
 
@@ -69,78 +71,77 @@ local function expiration_process(task, tuple)
     task.checked_tuples_count = task.checked_tuples_count + 1
     if task.is_tuple_expired(task.args, tuple) then
         task.expired_tuples_count = task.expired_tuples_count + 1
-        task.process_expired_tuple(task.space_id, task.args, tuple)
+        task.process_expired_tuple(task.space_id, task.args, tuple, task)
     end
 end
 
 -- stop for some time
-local function suspend_basic(scan_space, task, len)
+local function suspend_basic(task, len)
     local delay = (task.tuples_per_iteration * task.full_scan_time)
     delay = math.min(delay / len, task.iteration_delay)
     fiber.sleep(delay)
 end
 
-local function suspend(scan_space, task)
+local function suspend(task)
     -- Return the number of tuples in the space
-    local space_len = scan_space:len()
+    local space_len = task.expire_index:len()
     if space_len > 0 then
-        suspend_basic(scan_space, task, space_len)
+        suspend_basic(task, space_len)
     end
 end
 
 -- delete with some suspend and some tuples limit
-local function tree_index_iter(scan_space, task)
+local function tree_index_iter(task)
     -- iteration with GT iterator
-    local params = {iterator = 'GT', limit = task.tuples_per_iteration}
+    local params = {iterator = task.iterator, limit = task.tuples_per_iteration}
     local last_id
-    local tuples = scan_space.index[0]:select({}, params)
+    local tuples = task.expire_index:select(task.start_element, params)
     while #tuples > 0 do
         last_id = tuples[#tuples]
         for _, tuple in ipairs(tuples) do
+            if task:stop_iteration() then goto done end
             expiration_process(task, tuple)
         end
-        suspend(scan_space, task)
-        local key = construct_key(scan_space.id, last_id)
+        suspend(task)
+        local key = construct_key(task.expire_index, last_id)
         -- select all greater then last key
-        tuples = scan_space.index[0]:select(key, params)
+        tuples = task.expire_index:select(key, params)
     end
-
+    ::done::
 end
 
-local function hash_index_iter(scan_space, task)
+local function hash_index_iter(task)
     -- iteration for hash index
     local checked_tuples_count = 0
-    for _, tuple in scan_space.index[0]:pairs(nil, {iterator = box.index.ALL}) do
+    for _, tuple in task.expire_index:pairs(nil, {iterator = box.index.ALL}) do
         checked_tuples_count = checked_tuples_count + 1
+        if task:stop_iteration() then break end
         expiration_process(task, tuple)
         -- find out if the worker can go to sleep
         if checked_tuples_count >= task.tuples_per_iteration then
             checked_tuples_count = 0
-            suspend(scan_space, task)
+            suspend(task)
         end
     end
 end
 
 local function default_do_worker_iteration(task)
-    local scan_space = box.space[task.space_id]
-    local index_type = scan_space.index[0].type
+    local index_type = task.expire_index.type
 
     -- full index scan loop
     if index_type == 'HASH' then
-        hash_index_iter(scan_space, task)
+        hash_index_iter(task)
     else
-        tree_index_iter(scan_space, task)
+        tree_index_iter(task)
     end
 end
 
 local function vinyl_do_worker_iteration(task)
-    local scan_space = box.space[task.space_id]
-
     local checked_tuples_count = 0
     local space_len = task.vinyl_assumed_space_len
 
-    local params = {iterator = 'GT', limit = task.tuples_per_iteration}
-    local tuples = scan_space.index[0]:select({}, params)
+    local params = {iterator = task.iterator, limit = task.tuples_per_iteration}
+    local tuples = task.expire_index:select(task.start_element, params)
     while true do
         local tuple_cnt = #tuples
         if tuple_cnt == 0 then
@@ -149,16 +150,24 @@ local function vinyl_do_worker_iteration(task)
         local last_id = nil
         for _, tuple in ipairs(tuples) do
             last_id = tuple
+            if task:stop_iteration() then
+                checked_tuples_count = checked_tuples_count + tuple_cnt
+                if checked_tuples_count > space_len then
+                    space_len = task.vinyl_assumed_space_len_factor * space_len
+                end
+                goto done
+            end
             expiration_process(task, tuple)
         end
         checked_tuples_count = checked_tuples_count + tuple_cnt
         if checked_tuples_count > space_len then
             space_len = task.vinyl_assumed_space_len_factor * space_len
         end
-        local key = construct_key(scan_space.id, last_id)
-        suspend_basic(scan_space, task, space_len)
-        tuples = scan_space.index[0]:select(key, params)
+        local key = construct_key(task.expire_index, last_id)
+        suspend_basic(task, space_len)
+        tuples = task.expire_index:select(key, params)
     end
+    ::done::
     task.vinyl_assumed_space_len = checked_tuples_count
 end
 
@@ -267,6 +276,8 @@ local function create_task(name)
         is_tuple_expired      = nil,
         process_expired_tuple = nil,
         args                  = nil,
+        expire_index          = nil,
+        start_element         = nil,
         iteration_delay                = constants.max_delay,
         full_scan_delay                = constants.max_delay,
         tuples_per_iteration           = constants.default_tuples_per_iteration,
@@ -276,7 +287,9 @@ local function create_task(name)
         on_full_scan_error             = constants.default_on_full_scan,
         on_full_scan_success           = constants.default_on_full_scan,
         on_full_scan_start             = constants.default_on_full_scan,
-        on_full_scan_complete          = constants.default_on_full_scan
+        on_full_scan_complete          = constants.default_on_full_scan,
+        stop_iteration                 = constants.default_on_full_scan,
+        iterator                       = constants.iterator
     }, { __index = Task_methods })
     return task
 end
@@ -296,8 +309,9 @@ local function get_task(name)
 end
 
 -- default process_expired_tuple function
-local function default_tuple_drop(space_id, args, tuple)
-    box.space[space_id]:delete(construct_key(space_id, tuple))
+-- task as parameter was added to preserve backward compatibility
+local function default_tuple_drop(space_id, args, tuple, task)
+    task.expire_index:delete(construct_key(task.expire_index, tuple))
 end
 
 
@@ -315,6 +329,10 @@ end
 --   options = {      -- (table with named options)
 --     * process_expired_tuple -- applied to expired tuples, receives
 --                                (space_id, args, tuple) as arguments
+--     * index                 -- name of index, need to have an index on this name (default primary index)
+--     * ascending             -- false if descending (default true)
+--     * start_element         -- must be the same data type as index field or fields (default nil)
+--     * stop_iteration        -- call function before check is tuple expired, if false the iteration will be stopped
 --     * on_full_scan_start    -- call function on starting full scan iteration
 --     * on_full_scan_complete -- call function on complete full scan iteration
 --     * on_full_scan_success  -- call function on success full scan iteration
@@ -366,6 +384,88 @@ local function expirationd_run_task(name, space_id, is_tuple_expired, options)
     end
     task.process_expired_tuple = options.process_expired_tuple or default_tuple_drop
 
+    -- validate index field
+    --local expire_index
+    --if options.field then
+    --    if type(options.field) ~= "number" and type(options.field) ~= "string" then
+    --        error("field has a wrong type: " .. type(options.field))
+    --    end
+    --    local format_av = box.space[space_id]:format()
+    --    local format = {}
+    --    local have_format = false
+    --    -- add in format numbers of fields
+    --    -- it is needed to take field_no if field is string e.g. 'name'
+    --    for no, f in pairs(format_av) do
+    --        format[ f.name ] = {
+    --            name = f.name;
+    --            type = f.type;
+    --            no   = no;
+    --        }
+    --        format[ no ] = format[ f.name ];
+    --        have_format = true
+    --    end
+    --    for _, idx in pairs(box.space[space_id].index) do
+    --        for _, part in pairs(idx.parts) do
+    --            format[ part.fieldno ] = format[ part.fieldno ] or { no = part.fieldno }
+    --            format[ part.fieldno ].type = part.type
+    --        end
+    --    end
+    --
+    --    local expire_field_no
+    --    if have_format and type(options.field) == "string" then
+    --        expire_field_no = format[ options.field ].no
+    --    else
+    --        expire_field_no = options.field
+    --    end
+    --    if type(expire_field_no) ~= 'number' then
+    --        error("Need correct field option")
+    --    end
+    --
+    --    for _, index in pairs(box.space[space_id].index) do
+    --        -- we use only first part of index,
+    --        -- because we will have problems with the starting element
+    --        -- perhaps we should first of all take an index consisting only of our field
+    --        if index.parts[1].fieldno == expire_field_no then
+    --            expire_index = index
+    --        end
+    --    end
+    --    expire_index = expire_index
+    --else
+    --    -- take default primary index
+    --    expire_index = box.space[space_id].index[0]
+    --end
+    --task.expire_index = expire_index
+
+    -- validate index
+    local expire_index = box.space[space_id].index[0]
+    if options.index then
+        for _, index in pairs(box.space[space_id].index) do
+            if index.name == options.index then
+                expire_index = index
+            end
+        end
+    end
+    task.expire_index = expire_index
+
+    -- check ascending
+    if options.ascending ~= nil then
+        if type(options.ascending) ~= 'boolean' then
+            error("Invalid type of ascending")
+        end
+        if not options.ascending then
+            task.iterator = "LE"
+        end
+    end
+
+    task.start_element = options.start_element
+
+    if options.stop_iteration ~= nil then
+        if type(options.stop_iteration) ~= 'function' then
+            error("invalid type of stop_iteration is not function")
+        end
+        task.stop_iteration = options.stop_iteration
+    end
+
     -- check expire and process after expiration handler's arguments
     task.args = options.args