Support partial results (#118)

User may need only a subset of the query fields, but receives the values ​​of all fields, which leads to unnecessary network load. The solution is fields filtration on storage side by user-specified option. `fields` option allows to get only a subset of fields. This parameter contains the names of the fields that the user requests.

Author: AnaNek

CHANGELOG.md

@@ -16,6 +16,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
 
 ## [0.4.0] - 2020-12-02
 

README.md

@@ -71,6 +71,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns metadata and array contains one inserted row, error.
 
@@ -114,6 +115,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns metadata and array contains one row, error.
 
@@ -146,6 +148,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns metadata and array contains one updated row, error.
 
@@ -177,6 +180,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns metadata and array contains one deleted row (empty for vinyl), error.
 
@@ -210,6 +214,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns inserted or replaced rows and metadata or nil with error.
 
@@ -257,6 +262,7 @@ where:
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
   * `bucket_id` (`?number|cdata`) - bucket ID
+  * `fields` (`?table`) - field names for getting only a subset of fields
 
 Returns metadata and empty array of rows or nil, error.
 

crud/common/schema.lua

@@ -6,8 +6,10 @@ local errors = require('errors')
 local log = require('log')
 
 local ReloadSchemaError = errors.new_class('ReloadSchemaError',  {capture_stack = false})
+local FilterFieldsError = errors.new_class('FilterFieldsError',  {capture_stack = false})
 
 local const = require('crud.common.const')
+local dev_checks = require('crud.common.dev_checks')
 
 local schema = {}
 
@@ -133,22 +135,49 @@ local function get_space_schema_hash(space)
     return digest.murmur(msgpack.encode(space_info))
 end
 
+local function filter_result_fields(tuple, field_names)
+    if field_names == nil or tuple == nil then
+        return tuple
+    end
+
+    local result = {}
+
+    for i, field_name in ipairs(field_names) do
+        result[i] = tuple[field_name]
+        if result[i] == nil then
+            return nil, FilterFieldsError:new(
+                    'Space format doesn\'t contain field named %q', field_name
+            )
+        end
+    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 = ...}`
 -- space_schema_hash is computed if function failed and
 -- `add_space_schema_hash` is true
-function schema.wrap_box_space_func_result(add_space_schema_hash, space, func_name, ...)
+function schema.wrap_box_space_func_result(space, func_name, args, opts)
+    dev_checks('table', 'string', 'table', 'table')
+
     local result = {}
+    local err
 
-    local ok, func_res = pcall(space[func_name], space, ...)
+    opts = opts or {}
+
+    local ok, func_res = pcall(space[func_name], space, unpack(args))
     if not ok then
         result.err = func_res
-        if add_space_schema_hash then
+        if opts.add_space_schema_hash then
             result.space_schema_hash = get_space_schema_hash(space)
         end
     else
-        result.res = func_res
+        result.res, err = filter_result_fields(func_res, opts.field_names)
+        if err ~= nil then
+            return nil, err
+        end
     end
 
     return result

crud/common/utils.lua

@@ -10,9 +10,12 @@ local UnflattenError = errors.new_class("UnflattenError", {capture_stack = false
 local ParseOperationsError = errors.new_class('ParseOperationsError',  {capture_stack = false})
 local ShardingError = errors.new_class('ShardingError',  {capture_stack = false})
 local GetSpaceFormatError = errors.new_class('GetSpaceFormatError',  {capture_stack = false})
+local FilterFieldsError = errors.new_class('FilterFieldsError',  {capture_stack = false})
 
 local utils = {}
 
+local space_format_cache = setmetatable({}, {__mode = 'k'})
+
 function utils.table_count(table)
     dev_checks("table")
 
@@ -282,11 +285,57 @@ function utils.is_uuid(value)
     return ffi.istype(uuid_t, value)
 end
 
-function utils.format_result(rows, space)
-    return {
-        metadata = table.copy(space:format()),
-        rows = rows,
-    }
+local function get_field_format(space_format, field_name)
+    dev_checks('table', 'string')
+
+    local metadata = space_format_cache[space_format]
+    if metadata ~= nil then
+        return metadata[field_name]
+    end
+
+    space_format_cache[space_format] = {}
+    for _, field in ipairs(space_format) do
+        space_format_cache[space_format][field.name] = field
+    end
+
+    return space_format_cache[space_format][field_name]
+end
+
+local function filter_format_fields(space_format, field_names)
+    dev_checks('table', 'table')
+
+    local filtered_space_format = {}
+
+    for i, field_name in ipairs(field_names) do
+        filtered_space_format[i] = get_field_format(space_format, field_name)
+        if filtered_space_format[i] == nil then
+            return nil, FilterFieldsError:new(
+                    'Space format doesn\'t contain field named %q', field_name
+            )
+        end
+    end
+
+    return filtered_space_format
+end
+
+function utils.format_result(rows, space, field_names)
+    local result = {}
+    local err
+    local space_format = space:format()
+    result.rows = rows
+
+    if field_names == nil then
+        result.metadata = table.copy(space_format)
+        return result
+    end
+
+    result.metadata, err = filter_format_fields(space_format, field_names)
+
+    if err ~= nil then
+        return nil, err
+    end
+
+    return result
 end
 
 local function flatten_obj(space_name, obj)

crud/delete.lua

@@ -14,8 +14,8 @@ local delete = {}
 
 local DELETE_FUNC_NAME = '_crud.delete_on_storage'
 
-local function delete_on_storage(space_name, key)
-    dev_checks('string', '?')
+local function delete_on_storage(space_name, key, field_names)
+    dev_checks('string', '?', '?table')
 
     local space = box.space[space_name]
     if space == nil then
@@ -24,7 +24,10 @@ local function delete_on_storage(space_name, key)
 
     -- add_space_schema_hash is false because
     -- reloading space format on router can't avoid delete error on storage
-    return schema.wrap_box_space_func_result(false, space, 'delete', key)
+    return schema.wrap_box_space_func_result(space, 'delete', {key}, {
+        add_space_schema_hash = false,
+        field_names = field_names,
+    })
 end
 
 function delete.init()
@@ -38,6 +41,7 @@ local function call_delete_on_router(space_name, key, opts)
     dev_checks('string', '?', {
         timeout = '?number',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -54,7 +58,7 @@ local function call_delete_on_router(space_name, key, opts)
     local bucket_id = sharding.key_get_bucket_id(key, opts.bucket_id)
     local storage_result, err = call.rw_single(
         bucket_id, DELETE_FUNC_NAME,
-        {space_name, key},
+        {space_name, key, opts.fields},
         {timeout = opts.timeout}
     )
 
@@ -68,7 +72,7 @@ local function call_delete_on_router(space_name, key, opts)
 
     local tuple = storage_result.res
 
-    return utils.format_result({tuple}, space)
+    return utils.format_result({tuple}, space, opts.fields)
 end
 
 --- Deletes tuple from the specified space by key
@@ -96,6 +100,7 @@ function delete.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     return schema.wrap_func_reload(call_delete_on_router, space_name, key, opts)

crud/get.lua

@@ -14,8 +14,8 @@ local get = {}
 
 local GET_FUNC_NAME = '_crud.get_on_storage'
 
-local function get_on_storage(space_name, key)
-    dev_checks('string', '?')
+local function get_on_storage(space_name, key, field_names)
+    dev_checks('string', '?', '?table')
 
     local space = box.space[space_name]
     if space == nil then
@@ -24,7 +24,10 @@ local function get_on_storage(space_name, key)
 
     -- add_space_schema_hash is false because
     -- reloading space format on router can't avoid get error on storage
-    return schema.wrap_box_space_func_result(false, space, 'get', key)
+    return schema.wrap_box_space_func_result(space, 'get', {key}, {
+        add_space_schema_hash = false,
+        field_names = field_names,
+    })
 end
 
 function get.init()
@@ -38,6 +41,7 @@ local function call_get_on_router(space_name, key, opts)
     dev_checks('string', '?', {
         timeout = '?number',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -58,7 +62,7 @@ local function call_get_on_router(space_name, key, opts)
     -- a stale result.
     local storage_result, err = call.rw_single(
         bucket_id, GET_FUNC_NAME,
-        {space_name, key},
+        {space_name, key, opts.fields},
         {timeout = opts.timeout}
     )
 
@@ -77,7 +81,7 @@ local function call_get_on_router(space_name, key, opts)
         tuple = nil
     end
 
-    return utils.format_result({tuple}, space)
+    return utils.format_result({tuple}, space, opts.fields)
 end
 
 --- Get tuple from the specified space by key
@@ -105,6 +109,7 @@ function get.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
         bucket_id = '?number|cdata',
+        fields = '?table',
     })
 
     return schema.wrap_func_reload(call_get_on_router, space_name, key, opts)

crud/insert.lua

@@ -17,6 +17,7 @@ local INSERT_FUNC_NAME = '_crud.insert_on_storage'
 local function insert_on_storage(space_name, tuple, opts)
     dev_checks('string', 'table', {
         add_space_schema_hash = '?boolean',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -29,7 +30,10 @@ local function insert_on_storage(space_name, tuple, opts)
     -- add_space_schema_hash is true only in case of insert_object
     -- the only one case when reloading schema can avoid insert error
     -- is flattening object on router
-    return schema.wrap_box_space_func_result(opts.add_space_schema_hash, space, 'insert', tuple)
+    return schema.wrap_box_space_func_result(space, 'insert', {tuple}, {
+        add_space_schema_hash = opts.add_space_schema_hash,
+        field_names = opts.fields,
+    })
 end
 
 function insert.init()
@@ -44,6 +48,7 @@ local function call_insert_on_router(space_name, tuple, opts)
         timeout = '?number',
         bucket_id = '?number|cdata',
         add_space_schema_hash = '?boolean',
+        fields = '?table',
     })
 
     opts = opts or {}
@@ -60,6 +65,7 @@ local function call_insert_on_router(space_name, tuple, opts)
 
     local insert_on_storage_opts = {
         add_space_schema_hash = opts.add_space_schema_hash,
+        fields = opts.fields,
     }
 
     local storage_result, err = call.rw_single(
@@ -79,7 +85,7 @@ local function call_insert_on_router(space_name, tuple, opts)
 
     local tuple = storage_result.res
 
-    return utils.format_result({tuple}, space)
+    return utils.format_result({tuple}, space, opts.fields)
 end
 
 --- Inserts a tuple to the specified space
@@ -108,6 +114,7 @@ function insert.tuple(space_name, tuple, opts)
         timeout = '?number',
         bucket_id = '?number|cdata',
         add_space_schema_hash = '?boolean',
+        fields = '?table',
     })
 
     return schema.wrap_func_reload(call_insert_on_router, space_name, tuple, opts)