Some new features

Iterations over Secondary Indexes, ascending, start_element and
stop_iteration function

Author: ArtDu

Makefile

@@ -1,7 +1,9 @@
 all:
 	@echo "Only tests are available: make test"
 
+.PHONY: test
 test:
+	.rocks/bin/luatest -v
 	rm -rf *.xlog* *.snap 51{2,3,4,5,6,7}
 	INDEX_TYPE='TREE' SPACE_TYPE='vinyl' ./test.lua
 	rm -rf *.xlog* *.snap 51{2,3,4,5,6,7}

expirationd.lua

@@ -45,7 +45,13 @@ 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 function for process_while
+    default_process_while = function(...) return true end,
+    -- default iterating over the loop will go in ascending index
+    iterator = "GE",
+    -- default atomic_iteration is false, batch of items doesn't include in one transaction
+    atomic_iteration = false
 }
 
 -- ========================================================================= --
@@ -57,10 +63,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,95 +75,132 @@ 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_key, params)
     while #tuples > 0 do
         last_id = tuples[#tuples]
+        if task.atomic_iteration then
+            box.begin()
+        end
         for _, tuple in ipairs(tuples) do
+            if not task:process_while() then goto done end
             expiration_process(task, tuple)
         end
-        suspend(scan_space, task)
-        local key = construct_key(scan_space.id, last_id)
+        if task.atomic_iteration then
+            box.commit()
+        end
+        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, {iterator = 'GT', limit = task.tuples_per_iteration})
+    end
+    ::done::
+    if task.atomic_iteration then
+        box.commit()
     end
-
 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
+    if task.atomic_iteration then
+        box.begin()
+    end
+    for _, tuple in task.expire_index:pairs(nil, {iterator = box.index.ALL}) do
         checked_tuples_count = checked_tuples_count + 1
+        if not task:process_while() 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
+            if task.atomic_iteration then
+                box.commit()
+            end
             checked_tuples_count = 0
-            suspend(scan_space, task)
+            suspend(task)
+            if task.atomic_iteration then
+                box.begin()
+            end
         end
     end
+    if task.atomic_iteration then
+        box.commit()
+    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_key, params)
     while true do
         local tuple_cnt = #tuples
         if tuple_cnt == 0 then
             break
         end
         local last_id = nil
+        if task.atomic_iteration then
+            box.begin()
+        end
         for _, tuple in ipairs(tuples) do
             last_id = tuple
+            if not task:process_while() 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
+        if task.atomic_iteration then
+            box.commit()
+        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, {iterator = "GT", limit = task.tuples_per_iteration})
+    end
+    ::done::
+    if task.atomic_iteration then
+        box.commit()
     end
     task.vinyl_assumed_space_len = checked_tuples_count
 end
@@ -267,6 +310,8 @@ local function create_task(name)
         is_tuple_expired      = nil,
         process_expired_tuple = nil,
         args                  = nil,
+        expire_index          = nil,
+        start_key             = nil,
         iteration_delay                = constants.max_delay,
         full_scan_delay                = constants.max_delay,
         tuples_per_iteration           = constants.default_tuples_per_iteration,
@@ -276,7 +321,10 @@ 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,
+        process_while                  = constants.default_process_while,
+        iterator                       = constants.iterator,
+        atomic_iteration               = constants.atomic_iteration
     }, { __index = Task_methods })
     return task
 end
@@ -296,8 +344,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 +364,11 @@ 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)
+--     * atomic_iteration      -- boolean, true if we want put all items from batch in one transaction
+--     * iterator              -- iterator which can be used at the given index
+--     * start_key             -- must be the same data type as index field or fields (default nil)
+--     * process_while         -- 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,21 +420,108 @@ 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
+    local expire_index = box.space[space_id].index[0]
+    if options.index then
+        local index_exists = false
+        for _, index in pairs(box.space[space_id].index) do
+            if index.name == options.index then
+                expire_index = index
+                index_exists = true
+            end
+        end
+        if not index_exists then
+            error("Index with name " .. options.index .. " not exists")
+        end
+    end
+    task.expire_index = expire_index
+
+    -- check iterator
+    if options.iterator ~= nil then
+        if type(options.iterator) ~= 'string' then
+            error("Invalid type of iterator, expected string")
+        end
+        if expire_index.type == 'TREE' then
+            if  options.iterator ~= 'ALL' and
+                    options.iterator ~= 'EQ'  and
+                    options.iterator ~= 'REQ' and
+                    options.iterator ~= 'GT'  and
+                    options.iterator ~= 'GE'  and
+                    options.iterator ~= 'LT'  and
+                    options.iterator ~= 'LE'  then
+                error("Invalid iterator for TREE index, expected {ALL, EQ, REQ, GT, GE, LT, LE}")
+            end
+        elseif expire_index.type == 'HASH' then
+            if  options.iterator ~= 'ALL' and
+                    options.iterator ~= 'EQ'  and
+                    options.iterator ~= 'GT'  then
+                error("Invalid iterator for HASH index, expected {ALL, EQ, GT}")
+            end
+        end
+        task.iterator = options.iterator
+    else
+        if expire_index.type == 'HASH' then
+            task.iterator = "ALL"
+        end
+    end
+
+    -- check start_key
+    if options.start_key then
+        if type(options.start_key) == "table" then
+            if #expire_index.parts ~= #options.start_key then
+                error("Wrong number of elements in start_key table, expected: " .. #expire_index.parts)
+            end
+            for i = 1, #expire_index.parts do
+                log.info(expire_index.parts[i].type)
+                if expire_index.parts[i].type ~= type(options.start_key[i]) then
+                    error("Wrong type in " .. i ..
+                             " element of start_key table, expected: " .. expire_index.parts[i].type ..
+                             ", got: " .. type(options.start_key[i]))
+                end
+            end
+        else
+            if #expire_index.parts ~= 1 then
+                error("Wrong number of elements in start_key table, expected: " .. #expire_index.parts)
+            end
+            if expire_index.parts[1].type ~= type(options.start_key) then
+                error("Wrong type of start_key")
+            end
+        end
+        task.start_key = options.start_key
+    end
+
+
+    -- check process_while
+    if options.process_while ~= nil then
+        if type(options.process_while) ~= 'function' then
+            error("Invalid type of process_while is not function")
+        end
+        task.process_while = options.process_while
+    end
+
+    -- check transaction
+    if options.atomic_iteration ~= nil then
+        if type(options.atomic_iteration) ~= 'boolean' then
+            error("Invalid type of atomic_iteration, expected boolean")
+        end
+        task.atomic_iteration = options.atomic_iteration
+    end
+
     -- check expire and process after expiration handler's arguments
     task.args = options.args
 
     -- check tuples per iteration (not required)
     if options.tuples_per_iteration ~= nil then
         if options.tuples_per_iteration <= 0 then
-            error("invalid tuples per iteration parameter")
+            error("Invalid tuples per iteration parameter")
         end
         task.tuples_per_iteration = options.tuples_per_iteration
     end
 
     -- check full scan time
     if options.full_scan_time ~= nil then
         if options.full_scan_time <= 0 then
-            error("invalid full scan time")
+            error("Invalid full scan time")
         end
         task.full_scan_time = options.full_scan_time
     end

test.sh

@@ -11,5 +11,9 @@ deb-src https://packagecloud.io/tarantool/${VERSION}/ubuntu/ $release main
 EOF
 
 sudo apt-get update > /dev/null
-sudo apt-get -q -y install tarantool --force-yes
+sudo apt-get -q -y install tarantool tarantool-dev cartridge-cli npm --force-yes
+curl -sL https://deb.nodesource.com/setup_15.x | sudo bash -
+sudo apt-get install -y nodejs
+tarantoolctl rocks install luatest
+tarantoolctl rocks install cartridge
 make test