crud: add readview support

Added readview support for select and pairs.

Closes #343

crud.lua

@@ -20,6 +20,7 @@ local borders = require('crud.borders')
 local sharding_metadata = require('crud.common.sharding.sharding_metadata')
 local utils = require('crud.common.utils')
 local stats = require('crud.stats')
+local readview = require('crud.readview')
 
 local crud = {}
 
@@ -147,6 +148,8 @@ crud.reset_stats = stats.reset
 -- @function storage_info
 crud.storage_info = utils.storage_info
 
+crud.readview = readview.call
+
 --- Initializes crud on node
 --
 -- Exports all functions that are used for calls
@@ -174,6 +177,7 @@ function crud.init_storage()
     count.init()
     borders.init()
     sharding_metadata.init()
+    readview.init()
 
     _G._crud.storage_info_on_storage = utils.storage_info_on_storage
 end

crud/compare/filters.lua

@@ -146,6 +146,75 @@ local function parse(space, conditions, opts)
     return filter_conditions
 end
 
+local function parse_readview(space_f, conditions, opts)
+    dev_checks('table','table', '?table', {
+        scan_condition_num = '?number',
+        tarantool_iter = 'number',
+    })
+
+    conditions = conditions ~= nil and conditions or {}
+
+    local space_format = space_f:format()
+    local space_indexes = space_f.index
+
+    local fieldnos_by_names = {}
+
+    for i, field_format in ipairs(space_format) do
+        fieldnos_by_names[field_format.name] = i
+    end
+
+    local filter_conditions = {}
+    for i, condition in ipairs(conditions) do
+        if i ~= opts.scan_condition_num then
+            -- Index check (including one and multicolumn)
+            local fields
+            local fields_types = {}
+            local values_opts
+
+            local index = space_indexes[condition.operand]
+
+            if index ~= nil then
+                fields = get_index_fieldnos(index)
+                fields_types = get_index_fields_types(index)
+                values_opts = get_values_opts(index)
+            else
+                local fieldno = fieldnos_by_names[condition.operand]
+
+                if fieldno ~= nil then
+                    fields = {fieldno}
+                else
+                    -- We assume this is jsonpath, so it is
+                    -- not in fieldnos_by_name map.
+                    fields = {condition.operand}
+                end
+
+                local field_format = space_format[fieldno]
+                local is_nullable
+
+                if field_format ~= nil then
+                    fields_types = {field_format.type}
+                    is_nullable = field_format.is_nullable == true
+                end
+
+                values_opts = {
+                    {is_nullable = is_nullable, collation = nil},
+                }
+            end
+
+            table.insert(filter_conditions, {
+                fields = fields,
+                operator = condition.operator,
+                values = condition.values,
+                types = fields_types,
+                early_exit_is_possible = is_early_exit_possible(index, opts.tarantool_iter, condition),
+                values_opts = values_opts,
+            })
+        end
+    end
+
+    return filter_conditions
+end
+
 local function format_value(value)
     if type(value) == 'nil' then
         return 'nil'
@@ -665,6 +734,26 @@ function filters.gen_func(space, conditions, opts)
     return filter_func
 end
 
+function filters.gen_func_readview(space, space_format, conditions, opts)
+    dev_checks('table','table', '?table', {
+        tarantool_iter = 'number',
+        scan_condition_num = '?number',
+    })
+
+    local filter_conditions, err = parse_readview(space_format, conditions, {
+        scan_condition_num = opts.scan_condition_num,
+        tarantool_iter = opts.tarantool_iter,
+    })
+    if err ~= nil then
+        return nil, GenFiltersError:new("Failed to generate filters for specified conditions: %s", err)
+    end
+
+    local filter_code = gen_filter_code(filter_conditions)
+    local filter_func = compile(filter_code)
+
+    return filter_func
+end
+
 filters.internal = {
     parse = parse,
     gen_filter_code = gen_filter_code,

crud/compare/plan.lua

@@ -286,6 +286,131 @@ function plan.new(space, conditions, opts)
     return plan
 end
 
+function plan.new_readview(space, conditions, opts)
+    dev_checks('table', '?table', {
+        first = '?number',
+        after_tuple = '?table|cdata',
+        field_names = '?table',
+        force_map_call = '?boolean',
+        sharding_key_as_index_obj = '?table',
+        bucket_id = '?number|cdata',
+    })
+
+    conditions = conditions ~= nil and conditions or {}
+    opts = opts or {}
+
+    local space_name = space.name
+    local space_indexes = space.index
+    local space_format = space.format
+
+    if space_indexes == nil or next(space_indexes) == nil then
+        return nil, NoIndexesError:new('Space %q has no indexes, space should have primary index', space_name)
+    end
+
+    if conditions == nil then -- also cdata<NULL>
+        conditions = {}
+    end
+
+    local scan_index
+    local scan_iter
+    local scan_value
+    local scan_condition_num
+
+    local fieldno_map = utils.get_format_fieldno_map(space_format)
+
+    -- search index to iterate over
+    for i, condition in ipairs(conditions) do
+        scan_index = get_index_for_condition(space_indexes, space_format, condition)
+
+        if scan_index ~= nil then
+            scan_iter = compare_conditions.get_tarantool_iter(condition)
+            scan_value = condition.values
+            scan_condition_num = i
+            break
+        end
+    end
+
+    -- default iteration index is primary index
+    local primary_index = space_indexes[0]
+    if scan_index == nil then
+
+        if not index_is_allowed(primary_index) then
+            return nil, IndexTypeError:new('An index that matches specified conditions was not found: ' ..
+                'At least one of condition indexes or primary index should be of type TREE')
+        end
+
+        scan_index = primary_index
+        scan_iter = box.index.GE -- default iteration is `next greater than previous`
+        scan_value = {}
+    end
+
+    local cmp_key_parts = utils.merge_primary_key_parts(scan_index.parts, primary_index.parts)
+    local field_names = utils.enrich_field_names_with_cmp_key(opts.field_names, cmp_key_parts, space_format)
+
+    -- handle opts.first
+    local total_tuples_count
+    local scan_after_tuple, err = construct_after_tuple_by_fields(
+        fieldno_map, field_names, opts.after_tuple
+    )
+    if err ~= nil then
+        return nil, err
+    end
+
+    local is_equal_iter = scan_iter == box.index.EQ or scan_iter == box.index.REQ
+    if opts.first ~= nil then
+        total_tuples_count = math.abs(opts.first)
+
+        if opts.first < 0 then
+            scan_iter = utils.invert_tarantool_iter(scan_iter)
+
+            -- it makes no sence for EQ/REQ
+            if not is_equal_iter then
+                -- scan condition becomes border condition
+                scan_condition_num = nil
+
+                if scan_after_tuple ~= nil then
+                    -- after becomes a new scan value
+                    if has_keydef then
+                        local key_def = keydef_lib.new(scan_index.parts)
+                        scan_value = key_def:extract_key(scan_after_tuple)
+                    else
+                        scan_value = utils.extract_key(scan_after_tuple, scan_index.parts)
+                    end
+                else
+                    scan_value = nil
+                end
+            end
+        end
+    end
+
+    local sharding_key = nil
+    if opts.bucket_id == nil then
+        local sharding_index = opts.sharding_key_as_index_obj or primary_index
+
+        sharding_key = get_sharding_key_from_scan_value(scan_value, scan_index, scan_iter, sharding_index)
+
+        if sharding_key == nil then
+            sharding_key = extract_sharding_key_from_conditions(conditions, sharding_index,
+                                                                space_indexes, fieldno_map)
+        end
+    end
+
+    local plan = {
+        conditions = conditions,
+        space_name = space_name,
+        index_id = scan_index.id,
+        scan_value = scan_value,
+        after_tuple = scan_after_tuple,
+        scan_condition_num = scan_condition_num,
+        tarantool_iter = scan_iter,
+        total_tuples_count = total_tuples_count,
+        sharding_key = sharding_key,
+        field_names = field_names,
+    }
+
+    return plan
+end
+
 plan.internal = {
     get_sharding_key_from_scan_value = get_sharding_key_from_scan_value,
     extract_sharding_key_from_conditions = extract_sharding_key_from_conditions