select: first dirty merger integraion attempt

In scope of #33

Author: olegrok

crud/select.lua

@@ -2,14 +2,12 @@ local checks = require('checks')
 local errors = require('errors')
 local vshard = require('vshard')
 
-local call = require('crud.common.call')
 local registry = require('crud.common.registry')
 local utils = require('crud.common.utils')
 
 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')
@@ -22,9 +20,20 @@ local GetReplicasetsError = errors.new_class('GetReplicasetsError')
 local select_module = {}
 
 local SELECT_FUNC_NAME = '__select'
-
 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.limit,
+    }
+end
+
 local function call_select_on_storage(space_name, index_id, conditions, opts)
     checks('string', 'number', '?table', {
         scan_value = 'table',
@@ -63,7 +72,14 @@ local function call_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()
@@ -72,39 +88,6 @@ function select_module.init()
     })
 end
 
-local function select_iteration(space_name, plan, opts)
-    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(sharding_key)
     local bucket_id = vshard.router.bucket_id_strcrc32(sharding_key)
     local replicaset, err = vshard.router.route(bucket_id)
@@ -131,8 +114,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
-
     if opts.limit ~= nil and opts.limit < 0 then
         return nil, SelectError:new("limit should be >= 0")
     end
@@ -173,34 +154,21 @@ local function build_select_iterator(space_name, user_conditions, opts)
     -- set after tuple
     local after_tuple = utils.flatten(opts.after, space_format)
 
-    -- 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,
+    local select_opts = {
+        scan_value = plan.scan_value,
         after_tuple = after_tuple,
+        iter = plan.iter,
+        limit = opts.batch_size or DEFAULT_BATCH_SIZE,
+        scan_condition_num = plan.scan_condition_num,
+    }
 
-        batch_size = batch_size,
-        replicasets = replicasets_to_select,
+    local iter = Iterator.new(replicasets_to_select,
+            space_name, plan.index_id, plan.conditions, select_opts)
 
-        timeout = opts.timeout,
-    })
-
-    return iter
+    return {
+        iter = iter,
+        space_format = space_format,
+    }
 end
 
 function select_module.pairs(space_name, user_conditions, opts)
@@ -224,20 +192,23 @@ 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, _
+    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 obj, err = iter:get()
+        local obj, err = utils.unflatten(tuple, iter.space_format)
         if err ~= nil then
             error(string.format("Failed to get next object: %s", err))
         end
 
         return iter, obj
     end
 
-    return gen, nil, iter
+    return gen
 end
 
 function select_module.call(space_name, user_conditions, opts)
@@ -263,17 +234,18 @@ function select_module.call(space_name, user_conditions, opts)
 
     local objects = {}
 
-    while iter:has_next() do
-        local obj, err = iter:get()
-        if err ~= nil then
-            return nil, SelectError:new("Failed to get next object: %s", err)
-        end
-
-        if obj == nil then
+    local count = 0
+    for _, tuple in iter.iter:pairs() do
+        if opts.limit ~= nil and count >= opts.limit then
             break
         end
 
-        table.insert(objects, obj)
+        local object, err = utils.unflatten(tuple, iter.space_format)
+        if err ~= nil then
+            return nil, err
+        end
+        table.insert(objects, object)
+        count = count + 1
     end
 
     return objects