Iterate over any index

ArtDu · ArtDu · commit e2bb86a1c8f8 · 2021-07-05T13:08:20.000+03:00
Added the ability to iterate over any index by specifying the index name in options. The default is primary index. ci: installation, caching and running luatest PHONY added to Makefile as makefile target and luatests folder are the same. Needed for: #50
diff --git a/.github/workflows/fast_testing.yaml b/.github/workflows/fast_testing.yaml
@@ -35,4 +35,16 @@ jobs:
       - name: Clone the module
         uses: actions/checkout@v2
 
+      - name: Cache rocks
+        uses: actions/cache@v2
+        id: cache-rocks
+        with:
+          path: .rocks/
+          key: cache-rocks-${{ matrix.tarantool }}-01
+
+      - name: Install requirements
+        run: |
+          tarantoolctl rocks install luatest 0.5.0
+        if: steps.cache-rocks.outputs.cache-hit != 'true'
+
       - run: make test
diff --git a/Makefile b/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}
diff --git a/README.md b/README.md
@@ -40,6 +40,9 @@ Run a scheduled task to check and process (expire) tuples in a given space.
 * `options` -- (table with named options, may be nil)
     * `process_expired_tuple` - Applied to expired tuples, receives (space_id, args, tuple) as arguments.
      Can be nil: by default, tuples are removed.
+    * `index` - Name or id of the index to iterate on. If omitted, will use the primary index.
+     If there's no index with this name, will throw an error.
+     Supported index types are TREE and HASH, using other types will result in an error.
     * `tuples_per_iteration` - Number of tuples to check in one batch (iteration). Default is 1024.
     * `on_full_scan_start` - Function to call before starting a tuple scan.
     * `on_full_scan_complete` - Function to call after completing a full scan.
diff --git a/expirationd.lua b/expirationd.lua
@@ -69,7 +69,7 @@ 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
 
@@ -80,39 +80,38 @@ local function suspend_basic(task, len)
     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.index:len()
     if space_len > 0 then
         suspend_basic(task, space_len)
     end
 end
 
 local function default_do_worker_iteration(task)
     -- full index scan loop
-    local scan_space = box.space[task.space_id]
     local space_len = task.vinyl_assumed_space_len
     local checked_tuples_count = 0
     local vinyl_checked_tuples_count = 0
-    for _, tuple in scan_space.index[0]:pairs(nil, {iterator = box.index.ALL}) do
+    for _, tuple in task.index:pairs(nil, {iterator = box.index.ALL}) do
         checked_tuples_count = checked_tuples_count + 1
         vinyl_checked_tuples_count = vinyl_checked_tuples_count + 1
         expiration_process(task, tuple)
         -- find out if the worker can go to sleep
         -- if the batch is full
         if checked_tuples_count >= task.tuples_per_iteration then
             checked_tuples_count = 0
-            if scan_space.engine == "vinyl" then
+            if box.space[task.space_id].engine == "vinyl" then
                 if vinyl_checked_tuples_count > space_len then
                     space_len = task.vinyl_assumed_space_len_factor * space_len
                 end
                 suspend_basic(task, space_len)
             else
-                suspend(scan_space, task)
+                suspend(task)
             end
         end
     end
-    if scan_space.engine == "vinyl" then
+    if box.space[task.space_id].engine == "vinyl" then
         task.vinyl_assumed_space_len = vinyl_checked_tuples_count
     end
 end
@@ -222,6 +221,7 @@ local function create_task(name)
         is_tuple_expired      = nil,
         process_expired_tuple = nil,
         args                  = nil,
+        index                 = nil,
         iteration_delay                = constants.max_delay,
         full_scan_delay                = constants.max_delay,
         tuples_per_iteration           = constants.default_tuples_per_iteration,
@@ -271,6 +271,9 @@ end
 --   options = {      -- (table with named options)
 --     * process_expired_tuple -- Applied to expired tuples, receives (space_id, args, tuple) as arguments;
 --                                can be nil: by default, tuples are removed.
+--     * index                 -- Name or id of the index to iterate on; if omitted, will use the primary index;
+--                                if there's no index with this name, will throw an error.
+--                                supported index types are TREE and HASH, using other types will result in an error.
 --     * on_full_scan_start    -- Function to call before starting a full scan iteration.
 --     * on_full_scan_complete -- Function to call after completing a full scan iteration.
 --     * on_full_scan_success  -- Function to call after successfully completing a full scan iteration.
@@ -322,6 +325,25 @@ 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
+        if box.space[space_id].index[options.index] == nil then
+            if type(options.index) == "string" then
+                error("Index with name " .. options.index .. " does not exist")
+            elseif type(options.index) == "number" then
+                error("Index with id " .. options.index .. " does not exist")
+            else
+                error("Invalid type of index, expected string or number")
+            end
+        end
+        expire_index = box.space[space_id].index[options.index]
+        if expire_index.type ~= "TREE" and expire_index.type ~= "HASH" then
+            error("Not supported index type, expected TREE or HASH")
+        end
+    end
+    task.index = expire_index
+
     -- check expire and process after expiration handler's arguments
     task.args = options.args
 
diff --git a/test/helper.lua b/test/helper.lua
@@ -0,0 +1,111 @@
+local t = require("luatest")
+local fio = require("fio")
+
+local helpers = require("luatest.helpers")
+
+t.before_suite(function()
+    t.datadir = fio.tempdir()
+    box.cfg{
+        wal_dir    = t.datadir,
+        memtx_dir  = t.datadir,
+        vinyl_dir  = t.datadir,
+    }
+
+    local tree = box.schema.create_space("tree", { if_not_exists = true })
+    tree:format({
+        {name = "id", type = "number"},
+        {name = "first_name", type = "string"},
+        {name = "value", type = "number", is_nullable = true},
+        {name = "count", type = "number", is_nullable = true},
+        {name = "non_unique_id", type = "number", is_nullable = true, unique = false},
+        {name = "json_path_field", is_nullable = true, unique = false},
+        {name = "multikey_field", is_nullable = true},
+        {name = "functional_field", is_nullable = true},
+    })
+    tree:create_index("primary", {type = "TREE", parts={ 1 }})
+    tree:create_index("index_for_first_name", {type = "TREE", parts={ 2 }})
+    tree:create_index("multipart_index", {type = "TREE", parts={ {3, is_nullable = true}, {4, is_nullable = true} }})
+    tree:create_index("non_unique_index", {type = "TREE", parts={ {5, is_nullable = true} }, unique = false})
+
+    local hash = box.schema.create_space("hash", { if_not_exists = true })
+    hash:format({
+        {name = "id", type = "number"},
+        {name = "first_name", type = "string"},
+        {name = "value", type = "number", is_nullable = true},
+        {name = "count", type = "number", is_nullable = true}
+    })
+    hash:create_index("primary", {type = "HASH", parts={ 1 }} )
+    hash:create_index("index_for_first_name", {type = "HASH", parts={ 2 }} )
+    hash:create_index("multipart_index", {type = "HASH", parts={ {1}, {2} }})
+
+    local vinyl = box.schema.create_space("vinyl", { if_not_exists = true, engine = "vinyl" })
+    vinyl:format({
+        {name = "id", type = "number"},
+        {name = "first_name", type = "string"},
+        {name = "value", type = "number", is_nullable = true},
+        {name = "count", type = "number", is_nullable = true},
+        {name = "non_unique_id", type = "number", is_nullable = true, unique = false},
+        {name = "json_path_field", is_nullable = true, unique = false},
+        {name = "multikey_field", is_nullable = true, unique = false},
+    })
+    vinyl:create_index("primary", {type = "TREE", parts={ 1 }})
+    vinyl:create_index("index_for_first_name", {type = "TREE", parts={ 2 }})
+    vinyl:create_index("multipart_index", {type = "TREE", parts={ {3, is_nullable = true}, {4, is_nullable = true} }})
+    vinyl:create_index("non_unique_index", {type = "TREE", parts={ {5} }, unique = false })
+
+    if _TARANTOOL >= "2" then
+        local tree_code = [[function(tuple)
+            if tuple[8] then
+                return {string.sub(tuple[8],2,2)}
+            end
+            return {tuple[2]}
+        end]]
+        box.schema.func.create("tree_func",
+                {body = tree_code, is_deterministic = true, is_sandboxed = true})
+        tree:create_index("json_path_index",
+                {type = "TREE", parts = { {6, type = "scalar", path = "age", is_nullable = true} }})
+        tree:create_index("multikey_index",
+                {type = "TREE", parts = { {7, type = "str", path = "data[*].name"} }} )
+        tree:create_index("functional_index",
+                {type = "TREE", parts={ {1, type = "string"} }, func = "tree_func"})
+
+        vinyl:create_index("json_path_index",
+                {type = "TREE", parts = { {6, type = "scalar", path = "age", is_nullable = true} }})
+        vinyl:create_index("multikey_index",
+                {type = "TREE", parts = { {7, type = "str", path = "data[*].name", is_nullable = true} }})
+    end
+
+    local bitset = box.schema.create_space("bitset", { if_not_exists = true })
+    bitset:create_index("primary", {type = "TREE", parts={ 1 }})
+    bitset:create_index("index_for_first_name",
+            {type = "BITSET", parts={ {field = 2, type = "string"} }, unique = false})
+end)
+
+t.after_suite(function()
+    fio.rmtree(t.datadir)
+end)
+
+function helpers.init_spaces(g)
+    g.tree = box.space.tree
+    g.hash = box.space.hash
+    g.vinyl = box.space.vinyl
+    g.bitset = box.space.bitset
+end
+
+function helpers.truncate_spaces(g)
+    g.tree:truncate()
+    g.hash:truncate()
+    g.vinyl:truncate()
+end
+
+function helpers.is_expired_true()
+    return true
+end
+
+helpers.iteration_result = {}
+function helpers.is_expired_debug(_, tuple)
+    table.insert(helpers.iteration_result, tuple)
+    return true
+end
+
+return helpers
diff --git a/test/unit/custom_index_test.lua b/test/unit/custom_index_test.lua
