Add ref/unref for inser/update/upsert

Author: Satbek

crud/insert.lua

@@ -7,6 +7,7 @@ local utils = require('crud.common.utils')
 local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 local schema = require('crud.common.schema')
+local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')
 
 local InsertError = errors.new_class('InsertError', {capture_stack = false})
 
@@ -42,15 +43,32 @@ local function insert_on_storage(space_name, tuple, opts)
         return nil, err
     end
 
-    -- 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(space, 'insert', {tuple}, {
-        add_space_schema_hash = opts.add_space_schema_hash,
-        field_names = opts.fields,
-        noreturn = opts.noreturn,
-        fetch_latest_metadata = opts.fetch_latest_metadata,
-    })
+    local function make_insert()
+        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
+        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(bucket_id)
+        if not ref_ok then
+            return nil, err_ref
+        end
+
+        -- 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
+        local result = schema.wrap_box_space_func_result(space, 'insert', {tuple}, {
+            add_space_schema_hash = opts.add_space_schema_hash,
+            field_names = opts.fields,
+            noreturn = opts.noreturn,
+            fetch_latest_metadata = opts.fetch_latest_metadata,
+        })
+
+        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
+        if not unref_ok then
+            return nil, err_unref
+        end
+
+        return result
+    end
+
+    return box.atomic(make_insert)
 end
 
 insert.storage_api = {[INSERT_FUNC_NAME] = insert_on_storage}

crud/update.lua

@@ -9,6 +9,7 @@ local sharding_key_module = require('crud.common.sharding.sharding_key')
 local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
 local dev_checks = require('crud.common.dev_checks')
 local schema = require('crud.common.schema')
+local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')
 
 local UpdateError = errors.new_class('UpdateError', {capture_stack = false})
 
@@ -19,6 +20,7 @@ local CRUD_UPDATE_FUNC_NAME = utils.get_storage_call(UPDATE_FUNC_NAME)
 
 local function update_on_storage(space_name, key, operations, field_names, opts)
     dev_checks('string', '?', 'table', '?table', {
+        bucket_id = 'number|cdata',
         sharding_key_hash = '?number',
         sharding_func_hash = '?number',
         skip_sharding_hash_check = '?boolean',
@@ -42,39 +44,45 @@ local function update_on_storage(space_name, key, operations, field_names, opts)
         return nil, err
     end
 
-    -- add_space_schema_hash is false because
-    -- reloading space format on router can't avoid update error on storage
-    local res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
-        add_space_schema_hash = false,
-        field_names = field_names,
-        noreturn = opts.noreturn,
-        fetch_latest_metadata = opts.fetch_latest_metadata,
-    })
-
-    if err ~= nil then
-        return nil, err
-    end
-
-    if res.err == nil then
-        return res, nil
-    end
+    local function make_update()
+        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(opts.bucket_id)
+        if not ref_ok then
+            return nil, err_ref
+        end
 
-    -- Relevant for Tarantool older than 2.8.1.
-    -- We can only add fields to end of the tuple.
-    -- If schema is updated and nullable fields are added, then we will get error.
-    -- Therefore, we need to add filling of intermediate nullable fields.
-    -- More details: https://github.com/tarantool/tarantool/issues/3378
-    if utils.is_field_not_found(res.err.code) then
-        operations = utils.add_intermediate_nullable_fields(operations, space:format(), space:get(key))
-        res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
+        -- add_space_schema_hash is false because
+        -- reloading space format on router can't avoid update error on storage
+        local res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
             add_space_schema_hash = false,
             field_names = field_names,
             noreturn = opts.noreturn,
             fetch_latest_metadata = opts.fetch_latest_metadata,
         })
+
+        if err == nil and res.err ~= nil and utils.is_field_not_found(res.err.code) then
+            -- Relevant for Tarantool older than 2.8.1.
+            -- We can only add fields to end of the tuple.
+            -- If schema is updated and nullable fields are added, then we will get error.
+            -- Therefore, we need to add filling of intermediate nullable fields.
+            -- More details: https://github.com/tarantool/tarantool/issues/3378
+            operations = utils.add_intermediate_nullable_fields(operations, space:format(), space:get(key))
+            res, err = schema.wrap_box_space_func_result(space, 'update', {key, operations}, {
+                add_space_schema_hash = false,
+                field_names = field_names,
+                noreturn = opts.noreturn,
+                fetch_latest_metadata = opts.fetch_latest_metadata,
+            })
+        end
+
+        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(opts.bucket_id)
+        if not unref_ok then
+            return nil, err_unref
+        end
+
+        return res, err
     end
 
-    return res, err
+    return box.atomic(make_update)
 end
 
 update.storage_api = {[UPDATE_FUNC_NAME] = update_on_storage}
@@ -148,6 +156,7 @@ local function call_update_on_router(vshard_router, space_name, key, user_operat
     sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)
 
     local update_on_storage_opts = {
+        bucket_id = bucket_id_data.bucket_id,
         sharding_func_hash = bucket_id_data.sharding_func_hash,
         sharding_key_hash = sharding_key_hash,
         skip_sharding_hash_check = skip_sharding_hash_check,

crud/upsert.lua

@@ -7,6 +7,7 @@ local utils = require('crud.common.utils')
 local sharding = require('crud.common.sharding')
 local dev_checks = require('crud.common.dev_checks')
 local schema = require('crud.common.schema')
+local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')
 
 local UpsertError = errors.new_class('UpsertError', { capture_stack = false})
 
@@ -41,13 +42,30 @@ local function upsert_on_storage(space_name, tuple, operations, opts)
         return nil, err
     end
 
-    -- add_space_schema_hash is true only in case of upsert_object
-    -- the only one case when reloading schema can avoid insert error
-    -- is flattening object on router
-    return schema.wrap_box_space_func_result(space, 'upsert', {tuple, operations}, {
-        add_space_schema_hash = opts.add_space_schema_hash,
-        fetch_latest_metadata = opts.fetch_latest_metadata,
-    })
+    local function make_upsert()
+        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
+        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(bucket_id)
+        if not ref_ok then
+            return nil, err_ref
+        end
+
+        -- add_space_schema_hash is true only in case of upsert_object
+        -- the only one case when reloading schema can avoid insert error
+        -- is flattening object on router
+        local result = schema.wrap_box_space_func_result(space, 'upsert', {tuple, operations}, {
+            add_space_schema_hash = opts.add_space_schema_hash,
+            fetch_latest_metadata = opts.fetch_latest_metadata,
+        })
+
+        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
+        if not unref_ok then
+            return nil, err_unref
+        end
+
+        return result
+    end
+
+    return box.atomic(make_upsert)
 end
 
 upsert.storage_api = {[UPSERT_FUNC_NAME] = upsert_on_storage}