schema: fix stale schema for metadata generation

Corrects using of an stale schema to generate metadata for
`crud.update`, `crud.insert`, `crud.insert_*`, `crud.replace`,
`crud.replace_*`, `crud.upsert`, `crud.upsert_*`, `crud.delete`,
`crud.max`, `crud.min`, `crud.select`, `crud.pairs` and `crud.get`.

If the implemented `fetch_latest_metadata` option is used,
it is guaranteed that the metadata will be up-to-date.
Before receiving the space format, a mismatch check will be
performed between the scheme version on all involved storage
and the scheme version in the net_box connection of the router.
In case of mismatch, the schema reload will be triggered.

Closes #236

Author: GRISHNOV

CHANGELOG.md

@@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   `upsert`, `upsert_object`, `upsert_many`, `upsert_object_many`,
   `update`, `delete`. (#267).
 
+### Fixed
+* Crud DML operations returning stale schema for metadata generation.
+  Now you may use `fetch_latest_metadata` flag to work with latest schema (#236).
+
 ## [1.1.1] - 24-03-23
 
 ### Changed

README.md

@@ -236,6 +236,10 @@ where:
     the sole responsibility of the developer
   * `noreturn` (`?boolean`) - suppress successfully processed tuple
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array contains one inserted row, error.
 
@@ -303,6 +307,10 @@ where:
     the sole responsibility of the developer
   * `noreturn` (`?boolean`) - suppress successfully processed tuples
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array with inserted rows, array of errors.
 Each error object can contain field `operation_data`.
@@ -441,6 +449,10 @@ where:
   * `vshard_router` (`?string|table`) - Cartridge vshard group name or
     vshard router instance. Set this parameter if your space is not
     a part of the default vshard cluster
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array contains one row, error.
 
@@ -480,6 +492,10 @@ where:
     a part of the default vshard cluster
   * `noreturn` (`?boolean`) - suppress successfully processed tuple
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array contains one updated row, error.
 
@@ -518,6 +534,10 @@ where:
     a part of the default vshard cluster
   * `noreturn` (`?boolean`) - suppress successfully processed tuple
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array contains one deleted row (empty for vinyl), error.
 
@@ -567,6 +587,10 @@ where:
     the sole responsibility of the developer
   * `noreturn` (`?boolean`) - suppress successfully processed tuple
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns inserted or replaced rows and metadata or nil with error.
 
@@ -634,6 +658,10 @@ where:
     the sole responsibility of the developer
   * `noreturn` (`?boolean`) - suppress successfully processed tuples
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array with inserted/replaced rows, array of errors.
 Each error object can contain field `operation_data`.
@@ -772,6 +800,10 @@ where:
     a part of the default vshard cluster
   * `noreturn` (`?boolean`) - suppress successfully processed tuple
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and empty array of rows or nil, error.
 
@@ -835,6 +867,10 @@ where:
     a part of the default vshard cluster
   * `noreturn` (`?boolean`) - suppress successfully processed tuples
     (first return value is `nil`). `false` by default
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 Returns metadata and array of errors.
 Each error object can contain field `operation_data`.
@@ -977,6 +1013,10 @@ where:
     a part of the default vshard cluster
   * `yield_every` (`?number`) - number of tuples processed on storage to yield after,
     `yield_every` should be > 0, default value is 1000
+  * `fetch_latest_metadata` (`?boolean`) - guarantees the
+    up-to-date metadata (space format) in first return value, otherwise 
+    it may not take into account the latest migration of the data format. 
+    Performance overhead is up to 15%. `false` by default
 
 
 Returns metadata and array of rows, error.

crud/borders.lua

@@ -16,8 +16,8 @@ local borders = {}
 local STAT_FUNC_NAME = '_crud.get_border_on_storage'
 
 
-local function get_border_on_storage(border_name, space_name, index_id, field_names)
-    dev_checks('string', 'string', 'number', '?table')
+local function get_border_on_storage(border_name, space_name, index_id, field_names, fetch_latest_metadata)
+    dev_checks('string', 'string', 'number', '?table', '?boolean')
 
     assert(border_name == 'min' or border_name == 'max')
 
@@ -38,6 +38,7 @@ local function get_border_on_storage(border_name, space_name, index_id, field_na
     return schema.wrap_func_result(space, get_index_border, {index}, {
         add_space_schema_hash = true,
         field_names = field_names,
+        fetch_latest_metadata = fetch_latest_metadata,
     })
 end
 
@@ -71,9 +72,10 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
         timeout = '?number',
         fields = '?table',
         vshard_router = '?string|table',
+        fetch_latest_metadata = '?boolean',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router, opts.timeout)
+    local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)
     if err ~= nil then
         return nil, BorderError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -108,9 +110,9 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
         replicasets = replicasets,
         timeout = opts.timeout,
     }
-    local results, err = call.map(vshard_router,
+    local results, err, storages_info = call.map(vshard_router,
         STAT_FUNC_NAME,
-        {border_name, space_name, index.id, field_names},
+        {border_name, space_name, index.id, field_names, opts.fetch_latest_metadata},
         call_opts
     )
 
@@ -154,6 +156,14 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
         end
     end
 
+    if opts.fetch_latest_metadata == true then
+        -- This option is temporary and is related to [1], [2].
+        -- [1] https://github.com/tarantool/crud/issues/236
+        -- [2] https://github.com/tarantool/crud/issues/361
+        space = utils.fetch_latest_metadata_when_map_storages(space, space_name, vshard_router, opts,
+                                                              storages_info, netbox_schema_version)
+    end
+
     local result = utils.format_result({res_tuple}, space, field_names)
 
     if opts.fields ~= nil then

crud/common/map_call_cases/base_postprocessor.lua

@@ -13,6 +13,7 @@ function BasePostprocessor:new(vshard_router)
         early_exit = false,
         errs = nil,
         vshard_router = vshard_router,
+        storage_info = {},
     }
 
     setmetatable(postprocessor, self)
@@ -52,6 +53,14 @@ function BasePostprocessor:collect(result_info, err_info)
         wrapper_args = '?table',
     })
 
+    if result_info.value ~= nil and type(result_info.value[1]) == 'table' then
+        if result_info.value[1].storage_info ~= nil then
+            self.storage_info[result_info.key] = {
+                replica_schema_version = result_info.value[1].storage_info.replica_schema_version
+            }
+        end
+    end
+
     local err = err_info.err
     if err == nil and result_info.value[1] == nil then
         err = result_info.value[2]
@@ -78,8 +87,9 @@ end
 --
 -- @return[1] table results
 -- @return[2] table errs
+-- @return[3] table storage_info
 function BasePostprocessor:get()
-    return self.results, self.errs
+    return self.results, self.errs, self.storage_info
 end
 
 return BasePostprocessor

crud/common/map_call_cases/batch_postprocessor.lua

@@ -39,6 +39,10 @@ function BatchPostprocessor:collect(result_info, err_info)
         wrapper_args = '?table',
     })
 
+    if result_info.value ~= nil then
+        self.storage_info[result_info.key] = {replica_schema_version = result_info.value[3]}
+    end
+
     local errs = {err_info.err}
     if err_info.err == nil then
         errs = result_info.value[2]

crud/common/schema.lua

@@ -51,7 +51,7 @@ end
 local reload_in_progress = {}
 local reload_schema_cond = {}
 
-local function reload_schema(vshard_router)
+function schema.reload_schema(vshard_router)
     local replicasets = vshard_router:routeall()
     local vshard_router_name = vshard_router.name
 
@@ -96,14 +96,16 @@ function schema.wrap_func_reload(vshard_router, func, ...)
             break
         end
 
-        local ok, reload_schema_err = reload_schema(vshard_router)
+        local ok, reload_schema_err = schema.reload_schema(vshard_router)
         if not ok then
             log.warn("Failed to reload schema: %s", reload_schema_err)
             break
         end
 
         i = i + 1
         if i > const.RELOAD_RETRIES_NUM then
+            local warn_msg = "Number of attempts to reload schema has been ended: %s"
+            log.warn(warn_msg, const.RELOAD_RETRIES_NUM)
             break
         end
     end
@@ -221,6 +223,19 @@ function schema.wrap_func_result(space, func, args, opts)
         end
     end
 
+    if opts.fetch_latest_metadata == true then
+        local replica_schema_version
+        if box.info.schema_version ~= nil then
+            replica_schema_version = box.info.schema_version
+        else
+            replica_schema_version = box.internal.schema_version()
+        end
+        result.storage_info = {
+            replica_uuid = box.info().uuid,
+            replica_schema_version = replica_schema_version,
+        }
+    end
+
     return result
 end
 

crud/common/stash.lua

@@ -20,12 +20,16 @@ local stash = {}
 -- @tfield string stats_metrics_registry
 --  Stash for metrics rocks statistics registry.
 --
+-- @tfield string select_module_compat_info
+--  Stash for select compatability version registry.
+--
 stash.name = {
     cfg = '__crud_cfg',
     stats_internal = '__crud_stats_internal',
     stats_local_registry = '__crud_stats_local_registry',
     stats_metrics_registry = '__crud_stats_metrics_registry',
     ddl_triggers = '__crud_ddl_spaces_triggers',
+    select_module_compat_info = '__select_module_compat_info',
 }
 
 --- Setup Tarantool Cartridge reload.