Use merger module to fetch and sort tuples on router

Before this patch we used hand-crafted heap to sort tuples on
router. This patch removes it and changes to "merger" module
written in C.

"Merger" is available since Tarantool 2.2. Later "tuple-merger"
was introduced it's a version that is compatible with some
Tarantool 1.10 versions. In fact support several Tarantool
versions is quite complex task. Some msgpack helpers is not
avaliable in Tarantool 1.10 to solve this problem small
compatibility lawer is implemented.
Also "tuple-keydef" doesn't support collation_id options and it's
not so for built-in "key_def" module. We need to normalize index
parts and change "collation_id" to "collation" when create
"key_def" instance.

Closes #33

Author: olegrok

.github/workflows/test_on_push.yaml

@@ -21,6 +21,7 @@ jobs:
         run: |
           curl -L https://tarantool.io/installer.sh | sudo VER=${{ matrix.tarantool-version }} bash
           sudo apt install -y tarantool-dev
+          tarantool --version
           ./deps.sh
 
       - name: Run linter
@@ -46,6 +47,7 @@ jobs:
           rm -f tarantool-enterprise-bundle-${{ matrix.bundle_version }}.tar.gz
           sudo cp tarantool-enterprise/tarantool /usr/bin/tarantool
           source tarantool-enterprise/env.sh
+          tarantool --version
           ./deps.sh
 
       - name: Run linter

CMakeLists.txt

@@ -28,3 +28,9 @@ install(
   DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/cartridge
   DESTINATION ${TARANTOOL_INSTALL_LUADIR}
 )
+
+# Don't include to rockspec as some Tarantool versions (e.g. 2.2 and 2.3)
+# don't have symbols required by "tuple-merger" and "tuple-keydef" modules.
+execute_process(
+    COMMAND bash "-c" "tarantoolctl rocks install tuple-merger 0.0.1"
+)

crud/common/heap.lua

@@ -2,15 +2,13 @@ local checks = require('checks')
 local errors = require('errors')
 local vshard = require('vshard')
 
-local call = require('crud.common.call')
 local utils = require('crud.common.utils')
 local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 
 local select_conditions = require('crud.select.conditions')
 local select_plan = require('crud.select.plan')
 local select_executor = require('crud.select.executor')
-local select_comparators = require('crud.select.comparators')
 local select_filters = require('crud.select.filters')
 
 local Iterator = require('crud.select.iterator')
@@ -20,10 +18,16 @@ local GetReplicasetsError = errors.new_class('GetReplicasetsError')
 
 local select_module = {}
 
-local SELECT_FUNC_NAME = '_crud.select_on_storage'
-
 local DEFAULT_BATCH_SIZE = 100
 
+local function make_cursor(data)
+    local last_tuple = data[#data]
+
+    return {
+        after_tuple = last_tuple,
+    }
+end
+
 local function select_on_storage(space_name, index_id, conditions, opts)
     dev_checks('string', 'number', '?table', {
         scan_value = 'table',
@@ -62,46 +66,20 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         return nil, SelectError:new("Failed to execute select: %s", err)
     end
 
-    return tuples
+    local cursor
+    if #tuples < opts.limit or opts.limit == 0 then
+        cursor = {is_end = true}
+    else
+        cursor = make_cursor(tuples)
+    end
+
+    return cursor, tuples
 end
 
 function select_module.init()
    _G._crud.select_on_storage = select_on_storage
 end
 
-local function select_iteration(space_name, plan, opts)
-    dev_checks('string', '?table', {
-        after_tuple = '?table',
-        replicasets = 'table',
-        timeout = '?number',
-        limit = 'number',
-    })
-
-    -- call select on storages
-    local storage_select_opts = {
-        scan_value = plan.scan_value,
-        after_tuple = opts.after_tuple,
-        iter = plan.iter,
-        limit = opts.limit,
-        scan_condition_num = plan.scan_condition_num,
-    }
-
-    local storage_select_args = {
-        space_name, plan.index_id, plan.conditions, storage_select_opts,
-    }
-
-    local results, err = call.ro(SELECT_FUNC_NAME, storage_select_args, {
-        replicasets = opts.replicasets,
-        timeout = opts.timeout,
-    })
-
-    if err ~= nil then
-        return nil, err
-    end
-
-    return results
-end
-
 local function get_replicasets_by_sharding_key(bucket_id)
     local replicaset, err = vshard.router.route(bucket_id)
     if replicaset == nil then
@@ -128,8 +106,6 @@ local function build_select_iterator(space_name, user_conditions, opts)
         return nil, SelectError:new("batch_size should be > 0")
     end
 
-    local batch_size = opts.batch_size or DEFAULT_BATCH_SIZE
-
     -- check conditions
     local conditions, err = select_conditions.parse(user_conditions)
     if err ~= nil then
@@ -165,33 +141,33 @@ local function build_select_iterator(space_name, user_conditions, opts)
         replicasets_to_select = get_replicasets_by_sharding_key(bucket_id)
     end
 
-    -- generate tuples comparator
-    local scan_index = space.index[plan.index_id]
-    local primary_index = space.index[0]
-    local cmp_key_parts = utils.merge_primary_key_parts(scan_index.parts, primary_index.parts)
-    local cmp_operator = select_comparators.get_cmp_operator(plan.iter)
-    local tuples_comparator, err = select_comparators.gen_tuples_comparator(
-        cmp_operator, cmp_key_parts
-    )
-    if err ~= nil then
-        return nil, SelectError:new("Failed to generate comparator function: %s", err)
+    -- If opts.batch_size is missed we should specify it to min(first, DEFAULT_BATCH_SIZE)
+    local batch_size
+    if opts.batch_size == nil then
+        if opts.first ~= nil and opts.first < DEFAULT_BATCH_SIZE then
+            batch_size = opts.first
+        else
+            batch_size = DEFAULT_BATCH_SIZE
+        end
+    else
+        batch_size = opts.batch_size
     end
 
-    local iter = Iterator.new({
-        space_name = space_name,
-        space_format = space_format,
-        iteration_func = select_iteration,
-        comparator = tuples_comparator,
-
-        plan = plan,
-
-        batch_size = batch_size,
-        replicasets = replicasets_to_select,
+    local select_opts = {
+        scan_value = plan.scan_value,
+        after_tuple = plan.after_tuple,
+        iter = plan.iter,
+        limit = batch_size,
+        scan_condition_num = plan.scan_condition_num,
+    }
 
-        timeout = opts.timeout,
-    })
+    local iter = Iterator.new(replicasets_to_select,
+            space_name, plan.index_id, plan.conditions, select_opts)
 
-    return iter
+    return {
+        iter = iter,
+        space_format = space_format,
+    }
 end
 
 function select_module.pairs(space_name, user_conditions, opts)
@@ -222,31 +198,32 @@ function select_module.pairs(space_name, user_conditions, opts)
         error(string.format("Failed to generate iterator: %s", err))
     end
 
-    local gen = function(_, iter)
-        if not iter:has_next() then
+    local tuple, _
+
+    if opts.use_tomap ~= true then
+        return iter.iter:pairs()
+    end
+
+    local merger_gen = iter.iter:pairs()
+    local gen = function()
+        _, tuple = merger_gen.gen(nil, merger_gen.state)
+        if tuple == nil then
             return nil
         end
 
-        local tuple, err = iter:get()
+        local result
+        result, err = utils.unflatten(tuple, iter.space_format)
         if err ~= nil then
-            error(string.format("Failed to get next object: %s", err))
-        end
-
-        local result = tuple
-        if opts.use_tomap == true then
-            result, err = utils.unflatten(tuple, iter.space_format)
-            if err ~= nil then
-                error(string.format("Failed to unflatten next object: %s", err))
-            end
+            error(string.format("Failed to unflatten next object: %s", err))
         end
 
         return iter, result
     end
 
-    return gen, nil, iter
+    return gen
 end
 
-function select_module.call(space_name, user_conditions, opts)
+local function select_module_call_xc(space_name, user_conditions, opts)
     checks('string', '?table', {
         after = '?table',
         first = '?number',
@@ -277,17 +254,15 @@ function select_module.call(space_name, user_conditions, opts)
 
     local tuples = {}
 
-    while iter:has_next() do
-        local tuple, err = iter:get()
-        if err ~= nil then
-            return nil, SelectError:new("Failed to get next object: %s", err)
-        end
-
-        if tuple == nil then
+    local count = 0
+    local first = opts.first and math.abs(opts.first)
+    for _, tuple in iter.iter:pairs() do
+        if first ~= nil and count >= first then
             break
         end
 
         table.insert(tuples, tuple)
+        count = count + 1
     end
 
     if opts.first ~= nil and opts.first < 0 then
@@ -300,4 +275,8 @@ function select_module.call(space_name, user_conditions, opts)
     }
 end
 
+function select_module.call(space_name, user_conditions, opts)
+    return SelectError:pcall(select_module_call_xc, space_name, user_conditions, opts)
+end
+
 return select_module