crud: fix passing errors from storages for merger

Before this patch, returning any errors from storages for merger
operations (`crud.select`, `crud.pairs`, `readview:select`,
`readview:pairs`) had resulted in non-informative
"Invalid merge source 0x556bb6885a00" errors. This patch fixes the issue
by replacing error return with error throw.

Part of #373

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   non-iterating indexes (#373).
 * Precision loss for decimal conditions in case of non-indexed fields or
   non-iterating indexes (#373).
+* Passing errors from storages for merger operations (`crud.select`,
+  `crud.pairs`, `readview:select`, `readview:pairs`) (#423).
 
 ## [1.4.3] - 05-02-24
 

crud/readview.lua

@@ -133,22 +133,22 @@ local function select_readview_on_storage(space_name, index_id, conditions, opts
     end
 
     if space_readview == nil then
-        return cursor, ReadviewError:new("Space %q doesn't exist", space_name)
+        return ReadviewError:assert(false, "Space %q doesn't exist", space_name)
     end
 
     local space = box.space[space_name]
     if space == nil then
-        return cursor, ReadviewError:new("Space %q doesn't exist", space_name)
+        return ReadviewError:assert(false, "Space %q doesn't exist", space_name)
     end
     space_readview.format = space:format()
 
     local index_readview = space_readview.index[index_id]
     if index_readview == nil then
-        return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)
+        return ReadviewError:assert(false, "Index with ID %s doesn't exist", index_id)
     end
     local index = space.index[index_id]
     if index == nil then
-        return cursor, ReadviewError:new("Index with ID %s doesn't exist", index_id)
+        return ReadviewError:assert(false, "Index with ID %s doesn't exist", index_id)
     end
 
     local _, err = sharding.check_sharding_hash(space_name,
@@ -179,7 +179,7 @@ local function select_readview_on_storage(space_name, index_id, conditions, opts
         readview_index = index_readview,
     })
     if err ~= nil then
-        return cursor, ReadviewError:new("Failed to execute select: %s", err)
+        return ReadviewError:assert(false, "Failed to execute select: %s", err)
     end
 
     if resp.tuples_fetched < opts.limit or opts.limit == 0 then

crud/select.lua

@@ -61,12 +61,12 @@ local function select_on_storage(space_name, index_id, conditions, opts)
 
     local space = box.space[space_name]
     if space == nil then
-        return cursor, SelectError:new("Space %q doesn't exist", space_name)
+        SelectError:assert(false, "Space %q doesn't exist", space_name)
     end
 
     local index = space.index[index_id]
     if index == nil then
-        return cursor, SelectError:new("Index with ID %s doesn't exist", index_id)
+        SelectError:assert(false, "Index with ID %s doesn't exist", index_id)
     end
 
     local _, err = sharding.check_sharding_hash(space_name,
@@ -83,7 +83,7 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         scan_condition_num = opts.scan_condition_num,
     })
     if err ~= nil then
-        return cursor, SelectError:new("Failed to generate tuples filter: %s", err)
+        return SelectError:assert(false, "Failed to generate tuples filter: %s", err)
     end
 
     -- execute select
@@ -95,7 +95,7 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         yield_every = opts.yield_every,
     })
     if err ~= nil then
-        return cursor, SelectError:new("Failed to execute select: %s", err)
+        return SelectError:assert(false, "Failed to execute select: %s", err)
     end
 
     if resp.tuples_fetched < opts.limit or opts.limit == 0 then

test/integration/pairs_readview_test.lua

@@ -895,11 +895,13 @@ local function read_impl(cg, space, conditions, opts)
         local status, resp = pcall(function()
             return rv:pairs(space, conditions, opts):totable()
         end)
-        t.assert(status, resp)
-
         rv:close()
 
-        return resp, nil
+        if status then
+            return resp, nil
+        else
+            return nil, resp
+        end
     end, {space, conditions, opts})
 end
 
@@ -919,3 +921,17 @@ for case_name_template, case in pairs(gh_373_types_cases) do
         case(g, read_impl)
     end
 end
+
+pgroup.before_test(
+    'test_pairs_merger_process_storage_error',
+    read_scenario.before_merger_process_storage_error
+)
+
+pgroup.test_pairs_merger_process_storage_error = function(g)
+    read_scenario.merger_process_storage_error(g, read_impl)
+end
+
+pgroup.after_test(
+    'test_pairs_merger_process_storage_error',
+    read_scenario.after_merger_process_storage_error
+)

test/integration/pairs_test.lua

@@ -904,9 +904,12 @@ local function read_impl(cg, space, conditions, opts)
         local status, resp = pcall(function()
             return crud.pairs(space, conditions, opts):totable()
         end)
-        t.assert(status, resp)
 
-        return resp, nil
+        if status then
+            return resp, nil
+        else
+            return nil, resp
+        end
     end, {space, conditions, opts})
 end
 
@@ -926,3 +929,17 @@ for case_name_template, case in pairs(gh_373_types_cases) do
         case(g, read_impl)
     end
 end
+
+pgroup.before_test(
+    'test_pairs_merger_process_storage_error',
+    read_scenario.before_merger_process_storage_error
+)
+
+pgroup.test_pairs_merger_process_storage_error = function(g)
+    read_scenario.merger_process_storage_error(g, read_impl)
+end
+
+pgroup.after_test(
+    'test_pairs_merger_process_storage_error',
+    read_scenario.after_merger_process_storage_error
+)

test/integration/read_scenario.lua

@@ -50,11 +50,12 @@ local function gh_418_read_with_secondary_noneq_index_condition(cg, read)
     -- iterator had erroneously expected tuples to be sorted by `last_login`
     -- index while iterating on `city` index. Before the issue had beed fixed,
     -- user had received only one record instead of two.
-    local result = read(cg,
+    local result, err = read(cg,
         'logins',
         {{'=', 'city', 'Tatsumi Port Island'}, {'<=', 'last_login', 42}},
         {bucket_id = PINNED_BUCKET_NO}
     )
+    t.assert_equals(err, nil)
 
     if type(result) == 'number' then -- crud.count
         t.assert_equals(result, 2)
@@ -546,8 +547,91 @@ for space_kind, space_option in pairs(datetime_condition_space_options) do
     end
 end
 
+
+local function before_merger_process_storage_error(cg)
+    helpers.call_on_storages(cg.cluster, function(server)
+        server:exec(function()
+            local space
+            if box.info.ro == false then
+                space = box.schema.space.create('speedy_gonzales', {if_not_exists = true})
+
+                space:format({
+                    {name = 'id', type = 'unsigned'},
+                    {name = 'bucket_id', type = 'unsigned'},
+                })
+
+                space:create_index('pk', {
+                    parts = {'id'},
+                    if_not_exists = true,
+                })
+
+                space:create_index('bucket_id', {
+                    parts = {'bucket_id'},
+                    unique = false,
+                    if_not_exists = true,
+                })
+            end
+
+            local real_select_impl = rawget(_G, '_crud').select_on_storage
+            rawset(_G, '_real_select_impl', real_select_impl)
+
+            local real_select_readview_impl = rawget(_G, '_crud').select_readview_on_storage
+            rawset(_G, '_real_select_readview_impl', real_select_readview_impl)
+
+            -- Drop right before select to cause storage-side error.
+            -- Work guaranteed only with mode = 'write'.
+            local function erroneous_select_impl(...)
+                if box.info.ro == false then
+                    space:drop()
+                end
+
+                return real_select_impl(...)
+            end
+            rawget(_G, '_crud').select_on_storage = erroneous_select_impl
+
+            -- Close right before select to cause storage-side error.
+            -- Work guaranteed only with mode = 'write'.
+            local function erroneous_select_readview_impl(space_name, index_id, conditions, opts)
+                local list = box.read_view.list()
+
+                for k,v in pairs(list) do
+                    if v.id == opts.readview_id then
+                        list[k]:close()
+                    end
+                end
+
+                return real_select_readview_impl(space_name, index_id, conditions, opts)
+            end
+            rawget(_G, '_crud').select_readview_on_storage = erroneous_select_readview_impl
+        end)
+    end)
+end
+
+local function merger_process_storage_error(cg, read)
+    local _, err = read(cg, 'speedy_gonzales', {{'==', 'id', 1}})
+    t.assert_not_equals(err, nil)
+
+    local err_msg = err.err or tostring(err)
+    t.assert_str_contains(err_msg, "Space \"speedy_gonzales\" doesn't exist")
+end
+
+local function after_merger_process_storage_error(cg)
+    helpers.call_on_storages(cg.cluster, function(server)
+        server:exec(function()
+            local real_select_impl = rawget(_G, '_real_select_impl')
+            rawget(_G, '_crud').select_on_storage = real_select_impl
+
+            local real_select_readview_impl = rawget(_G, '_real_select_readview_impl')
+            rawget(_G, '_crud').select_readview_on_storage = real_select_readview_impl
+        end)
+    end)
+end
+
 return {
     gh_418_read_with_secondary_noneq_index_condition = gh_418_read_with_secondary_noneq_index_condition,
     gh_373_read_with_decimal_condition_cases = gh_373_read_with_decimal_condition_cases,
     gh_373_read_with_datetime_condition_cases = gh_373_read_with_datetime_condition_cases,
+    before_merger_process_storage_error = before_merger_process_storage_error,
+    merger_process_storage_error = merger_process_storage_error,
+    after_merger_process_storage_error = after_merger_process_storage_error,
 }

test/integration/select_readview_test.lua

@@ -2493,11 +2493,13 @@ local function read_impl(cg, space, conditions, opts)
         t.assert_equals(err, nil)
 
         local resp, err = rv:select(space, conditions, opts)
-        t.assert_equals(err, nil)
-
         rv:close()
 
-        return crud.unflatten_rows(resp.rows, resp.metadata)
+        if err ~= nil then
+            return nil, err
+        end
+
+        return crud.unflatten_rows(resp.rows, resp.metadata), nil
     end, {space, conditions, opts})
 end
 
@@ -2517,3 +2519,17 @@ for case_name_template, case in pairs(gh_373_types_cases) do
         case(g, read_impl)
     end
 end
+
+pgroup.before_test(
+    'test_select_merger_process_storage_error',
+    read_scenario.before_merger_process_storage_error
+)
+
+pgroup.test_select_merger_process_storage_error = function(g)
+    read_scenario.merger_process_storage_error(g, read_impl)
+end
+
+pgroup.after_test(
+    'test_select_merger_process_storage_error',
+    read_scenario.after_merger_process_storage_error
+)

test/integration/select_test.lua

@@ -2273,9 +2273,12 @@ local function read_impl(cg, space, conditions, opts)
     opts.mode = 'write'
 
     local resp, err = cg.cluster.main_server:call('crud.select', {space, conditions, opts})
-    t.assert_equals(err, nil)
 
-    return crud.unflatten_rows(resp.rows, resp.metadata)
+    if err ~= nil then
+        return nil, err
+    end
+
+    return crud.unflatten_rows(resp.rows, resp.metadata), nil
 end
 
 pgroup.test_gh_418_select_with_secondary_noneq_index_condition = function(g)
@@ -2294,3 +2297,17 @@ for case_name_template, case in pairs(gh_373_types_cases) do
         case(g, read_impl)
     end
 end
+
+pgroup.before_test(
+    'test_select_merger_process_storage_error',
+    read_scenario.before_merger_process_storage_error
+)
+
+pgroup.test_select_merger_process_storage_error = function(g)
+    read_scenario.merger_process_storage_error(g, read_impl)
+end
+
+pgroup.after_test(
+    'test_select_merger_process_storage_error',
+    read_scenario.after_merger_process_storage_error
+)