Support partial results for select/pairs (#122)

User may need to select a subset of the query fields, but receives the values of all fields, which leads to unnecessary network load. Using `fields` parameters allows to reduce amount of data transferred from storage. Result contains only fields specified by `fields` parameter, but scan key and primary key values are merged to the result fields to support pagination.

Author: AnaNek

CHANGELOG.md

@@ -19,7 +19,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 * Support for UUID field types and UUID values
-* `fields` option for simple operations to get partial result
+* `fields` option for simple operations and select/pairs calls with pagination support to get partial result
 
 ## [0.4.0] - 2020-12-02
 

crud/common/schema.lua

@@ -135,7 +135,7 @@ local function get_space_schema_hash(space)
     return digest.murmur(msgpack.encode(space_info))
 end
 
-local function filter_result_fields(tuple, field_names)
+local function filter_tuple_fields(tuple, field_names)
     if field_names == nil or tuple == nil then
         return tuple
     end
@@ -154,6 +154,28 @@ local function filter_result_fields(tuple, field_names)
     return result
 end
 
+function schema.filter_tuples_fields(tuples, field_names)
+    dev_checks('?table', '?table')
+
+    if field_names == nil then
+        return tuples
+    end
+
+    local result = {}
+
+    for _, tuple in ipairs(tuples) do
+        local filtered_tuple, err = filter_tuple_fields(tuple, field_names)
+
+        if err ~= nil then
+            return nil, err
+        end
+
+        table.insert(result, filtered_tuple)
+    end
+
+    return result
+end
+
 -- schema.wrap_box_space_func_result pcalls some box.space function
 -- and returns its result as a table
 -- `{res = ..., err = ..., space_schema_hash = ...}`
@@ -174,7 +196,7 @@ function schema.wrap_box_space_func_result(space, func_name, args, opts)
             result.space_schema_hash = get_space_schema_hash(space)
         end
     else
-        result.res, err = filter_result_fields(func_res, opts.field_names)
+        result.res, err = filter_tuple_fields(func_res, opts.field_names)
         if err ~= nil then
             return nil, err
         end

crud/common/utils.lua

@@ -377,6 +377,22 @@ local function filter_format_fields(space_format, field_names)
     return filtered_space_format
 end
 
+function utils.get_fields_format(space_format, field_names)
+    dev_checks('table', '?table')
+
+    if field_names == nil then
+        return table.copy(space_format)
+    end
+
+    local filtered_space_format, err = filter_format_fields(space_format, field_names)
+
+    if err ~= nil then
+        return nil, err
+    end
+
+    return filtered_space_format
+end
+
 function utils.format_result(rows, space, field_names)
     local result = {}
     local err

crud/select.lua

@@ -33,6 +33,7 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         iter = 'number',
         limit = 'number',
         scan_condition_num = '?number',
+        field_names = '?table',
     })
 
     local space = box.space[space_name]
@@ -64,7 +65,9 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         return nil, SelectError:new("Failed to execute select: %s", err)
     end
 
-    return tuples
+    -- getting tuples with user defined fields (if `fields` option is specified)
+    -- and fields that are needed for comparison on router (primary key + scan key)
+    return schema.filter_tuples_fields(tuples, opts.field_names)
 end
 
 function select_module.init()
@@ -86,6 +89,7 @@ local function select_iteration(space_name, plan, opts)
         iter = plan.iter,
         limit = opts.limit,
         scan_condition_num = plan.scan_condition_num,
+        field_names = plan.field_names,
     }
 
     local storage_select_args = {
@@ -125,6 +129,7 @@ local function build_select_iterator(space_name, user_conditions, opts)
         timeout = '?number',
         batch_size = '?number',
         bucket_id = '?number|cdata',
+        field_names = '?table',
     })
 
     opts = opts or {}
@@ -157,6 +162,7 @@ local function build_select_iterator(space_name, user_conditions, opts)
     local plan, err = select_plan.new(space, conditions, {
         first = opts.first,
         after_tuple = opts.after,
+        field_names = opts.field_names,
     })
 
     if err ~= nil then
@@ -181,16 +187,27 @@ local function build_select_iterator(space_name, user_conditions, opts)
     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)
+
+    -- generator of tuples comparator needs field_names and space_format
+    -- to update fieldno in each part in cmp_key_parts because storage result contains
+    -- fields in order specified by field_names
     local tuples_comparator, err = select_comparators.gen_tuples_comparator(
-        cmp_operator, cmp_key_parts
+        cmp_operator, cmp_key_parts, opts.field_names, space_format
     )
     if err ~= nil then
         return nil, SelectError:new("Failed to generate comparator function: %s", err)
     end
 
+    -- filter space format by plan.field_names (user defined fields + primary key + scan key)
+    -- to pass it user as metadata
+    local filtered_space_format, err = utils.get_fields_format(space_format, plan.field_names)
+    if err ~= nil then
+        return nil, err
+    end
+
     local iter = Iterator.new({
         space_name = space_name,
-        space_format = space_format,
+        space_format = filtered_space_format,
         iteration_func = select_iteration,
         comparator = tuples_comparator,
 
@@ -213,6 +230,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         batch_size = '?number',
         use_tomap = '?boolean',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -227,6 +245,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         timeout = opts.timeout,
         batch_size = opts.batch_size,
         bucket_id = opts.bucket_id,
+        field_names = opts.fields,
     }
 
     local iter, err = schema.wrap_func_reload(
@@ -268,6 +287,7 @@ function select_module.call(space_name, user_conditions, opts)
         timeout = '?number',
         batch_size = '?number',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -284,6 +304,7 @@ function select_module.call(space_name, user_conditions, opts)
         timeout = opts.timeout,
         batch_size = opts.batch_size,
         bucket_id = opts.bucket_id,
+        field_names = opts.fields,
     }
 
     local iter, err = schema.wrap_func_reload(
@@ -313,7 +334,7 @@ function select_module.call(space_name, user_conditions, opts)
     end
 
     return {
-        metadata = table.copy(iter.space_format),
+        metadata = iter.space_format,
         rows = tuples,
     }
 end

crud/select/comparators.lua

@@ -83,6 +83,37 @@ local function gen_array_cmp_func(target, key_parts)
     end
 end
 
+-- if `fields` option is specified then tuples to compare contains
+-- fields in order specified by field_names
+-- for comparison tuples we need to change fieldno in key_parts according to field_names
+local function update_key_parts_by_field_names(space_format, field_names, key_parts)
+    if field_names == nil then
+        return key_parts
+    end
+
+    local fields_positions = {}
+    local updated_key_parts = {}
+    local last_position = #field_names + 1
+
+    for i, field_name in ipairs(field_names) do
+        fields_positions[field_name] = i
+    end
+
+    for _, part in ipairs(key_parts) do
+        local field_name = space_format[part.fieldno].name
+        if not fields_positions[field_name] then
+            fields_positions[field_name] = last_position
+            last_position = last_position + 1
+        end
+        local updated_part = {type = part.type,
+                              fieldno = fields_positions[field_name],
+                              is_nullable = part.is_nullable}
+        table.insert(updated_key_parts, updated_part)
+    end
+
+    return updated_key_parts
+end
+
 local cmp_operators_by_tarantool_iter = {
     [box.index.GT] = operators.GT,
     [box.index.GE] = operators.GT,
@@ -127,15 +158,18 @@ function comparators.gen_func(cmp_operator, key_parts)
     return func
 end
 
-function comparators.gen_tuples_comparator(cmp_operator, key_parts)
-    local keys_comparator, err = comparators.gen_func(cmp_operator, key_parts)
+function comparators.gen_tuples_comparator(cmp_operator, key_parts, field_names, space_format)
+    local updated_key_parts = update_key_parts_by_field_names(
+            space_format, field_names, key_parts
+    )
+    local keys_comparator, err = comparators.gen_func(cmp_operator, updated_key_parts)
     if err ~= nil then
         return nil, ComparatorsError:new("Failed to generate comparator function: %s", err)
     end
 
     return function(lhs, rhs)
-        local lhs_key = utils.extract_key(lhs, key_parts)
-        local rhs_key = utils.extract_key(rhs, key_parts)
+        local lhs_key = utils.extract_key(lhs, updated_key_parts)
+        local rhs_key = utils.extract_key(rhs, updated_key_parts)
 
         return keys_comparator(lhs_key, rhs_key)
     end

crud/select/iterator.lua

@@ -25,6 +25,7 @@ function Iterator.new(opts)
         replicasets = 'table',
 
         timeout = '?number',
+        field_names = '?table',
     })
 
     local iter = {
@@ -35,6 +36,7 @@ function Iterator.new(opts)
         plan = opts.plan,
 
         timeout = opts.timeout,
+        field_names = opts.field_names,
 
         replicasets = table.copy(opts.replicasets),
         replicasets_count = utils.table_count(opts.replicasets),
@@ -99,6 +101,7 @@ local function update_replicasets_tuples(iter, after_tuple, replicaset_uuid)
         replicasets = replicasets,
         timeout = iter.timeout,
         limit = limit_per_storage_call,
+        field_names = iter.field_names,
     })
     if err ~= nil then
         return false, UpdateTuplesError:new('Failed to select tuples from storages: %s', err)

crud/select/plan.lua

@@ -9,6 +9,7 @@ local select_plan = {}
 local SelectPlanError = errors.new_class('SelectPlanError', {capture_stack = false})
 local IndexTypeError = errors.new_class('IndexTypeError', {capture_stack = false})
 local ValidateConditionsError = errors.new_class('ValidateConditionsError', {capture_stack = false})
+local FilterFieldsError = errors.new_class('FilterFieldsError',  {capture_stack = false})
 
 local function index_is_allowed(index)
     return index.type == 'TREE'
@@ -106,10 +107,70 @@ local function extract_sharding_key_from_scan_value(scan_value, scan_index, shar
     return sharding_key
 end
 
+-- We need to construct after_tuple by field_names
+-- because if `fields` option is specified we have after_tuple with partial fields
+-- and these fields are ordered by field_names + primary key + scan key
+-- this order can be differ from order in space format
+-- so we need to cast after_tuple to space format for scrolling tuples on storage
+local function construct_after_tuple_by_fields(space_format, field_names, tuple)
+    if tuple == nil then
+        return nil
+    end
+
+    if field_names == nil then
+        return tuple
+    end
+
+    local positions = {}
+    local transformed_tuple = {}
+
+    for i, field in ipairs(space_format) do
+        positions[field.name] = i
+    end
+
+    for i, field_name in ipairs(field_names) do
+        local fieldno = positions[field_name]
+        if fieldno == nil then
+            return nil, FilterFieldsError:new(
+                    'Space format doesn\'t contain field named %q', field_name
+            )
+        end
+
+        transformed_tuple[fieldno] = tuple[i]
+    end
+
+    return transformed_tuple
+end
+
+local function enrich_field_names_with_cmp_key(field_names, key_parts, space_format)
+    if field_names == nil then
+        return nil
+    end
+
+    local enriched_field_names = {}
+    local key_field_names = {}
+
+    for _, field_name in ipairs(field_names) do
+        table.insert(enriched_field_names, field_name)
+        key_field_names[field_name] = true
+    end
+
+    for _, part in ipairs(key_parts) do
+        local field_name = space_format[part.fieldno].name
+        if not key_field_names[field_name] then
+            table.insert(enriched_field_names, field_name)
+            key_field_names[field_name] = true
+        end
+    end
+
+    return enriched_field_names
+end
+
 function select_plan.new(space, conditions, opts)
     dev_checks('table', '?table', {
         first = '?number',
         after_tuple = '?table',
+        field_names = '?table',
     })
 
     conditions = conditions ~= nil and conditions or {}
@@ -159,9 +220,17 @@ function select_plan.new(space, conditions, opts)
         scan_value = {}
     end
 
+    local cmp_key_parts = utils.merge_primary_key_parts(scan_index.parts, primary_index.parts)
+    local field_names = 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 = opts.after_tuple
+    local scan_after_tuple, err = construct_after_tuple_by_fields(
+            space_format, field_names, opts.after_tuple
+    )
+    if err ~= nil then
+        return nil, err
+    end
 
     if opts.first ~= nil then
         total_tuples_count = math.abs(opts.first)
@@ -203,6 +272,7 @@ function select_plan.new(space, conditions, opts)
         iter = scan_iter,
         total_tuples_count = total_tuples_count,
         sharding_key = sharding_key,
+        field_names = field_names,
     }
 
     return plan