select: first dirty merger integraion attempt

In scope of #33

Author: olegrok

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,20 @@ 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(opts, data)
+    local last_tuple = data[#data]
+
+    return {
+        scan_value = opts.scan_value,
+        after_tuple = last_tuple,
+        iter = opts.iter,
+        limit = opts.limit,
+        scan_condition_num = opts.scan_condition_num,
+    }
+end
+
 local function select_on_storage(space_name, index_id, conditions, opts)
     dev_checks('string', 'number', '?table', {
         scan_value = 'table',
@@ -62,46 +70,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(opts, 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 +110,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 +145,21 @@ 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)
-    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 = opts.batch_size or DEFAULT_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 +190,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 +246,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 +267,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