carpawell/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#188] metabase: Refactor object put to metabase

Browse source 
Resolve funlen linter for db.put method

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-03-30 13:42:10 +03:00
parent 8273a3dfb2
commit 0739c36a3b
1 changed files with 32 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
pkg/local_object_storage/metabase/put.go
View file

@ -10,6 +10,7 @@ import (
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object" objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
storagelog "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/log" storagelog "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/log"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"github.com/nspcc-dev/neo-go/pkg/io" "github.com/nspcc-dev/neo-go/pkg/io"
@ -76,10 +77,11 @@ func (db *DB) Put(prm PutPrm) (res PutRes, err error) {
return return
} }
// nolint: funlen func (db *DB) put(tx *bbolt.Tx,
func (db *DB) put( obj *objectSDK.Object,
tx *bbolt.Tx, obj *objectSDK.Object, id []byte, id []byte,
si *objectSDK.SplitInfo, currEpoch uint64) error { si *objectSDK.SplitInfo,
currEpoch uint64) error {
cnr, ok := obj.ContainerID() cnr, ok := obj.ContainerID()
if !ok { if !ok {
return errors.New("missing container in object") return errors.New("missing container in object")
@ -95,25 +97,34 @@ func (db *DB) put(
return err // return any error besides SplitInfoError return err // return any error besides SplitInfoError
} }
// most right child and split header overlap parent so we have to
// check if object exists to not overwrite it twice
if exists { if exists {
// When storage engine moves objects between different sub-storages, return db.updateObj(tx, obj, id, si, isParent)
// it calls metabase.Put method with new storage ID, thus triggering this code.
if !isParent && id != nil {
return updateStorageID(tx, object.AddressOf(obj), id)
}
// when storage already has last object in split hierarchy and there is
// a linking object to put (or vice versa), we should update split info
// with object ids of these objects
if isParent {
return updateSplitInfo(tx, object.AddressOf(obj), si)
}
return nil
} }
return db.insertObject(tx, obj, id, si, isParent, cnr, currEpoch)
}
func (db *DB) updateObj(tx *bbolt.Tx, obj *objectSDK.Object, id []byte, si *objectSDK.SplitInfo, isParent bool) error {
// most right child and split header overlap parent so we have to
// check if object exists to not overwrite it twice
// When storage engine moves objects between different sub-storages,
// it calls metabase.Put method with new storage ID, thus triggering this code.
if !isParent && id != nil {
return updateStorageID(tx, object.AddressOf(obj), id)
}
// when storage already has last object in split hierarchy and there is
// a linking object to put (or vice versa), we should update split info
// with object ids of these objects
if isParent {
return updateSplitInfo(tx, object.AddressOf(obj), si)
}
return nil
}
func (db *DB) insertObject(tx *bbolt.Tx, obj *objectSDK.Object, id []byte, si *objectSDK.SplitInfo, isParent bool, cnr cid.ID, currEpoch uint64) error {
if par := obj.Parent(); par != nil && !isParent { // limit depth by two if par := obj.Parent(); par != nil && !isParent { // limit depth by two
parentSI, err := splitInfoFromObject(obj) parentSI, err := splitInfoFromObject(obj)
if err != nil { if err != nil {
@ -126,7 +137,7 @@ func (db *DB) put(
} }
} }
err = putUniqueIndexes(tx, obj, si, id) err := putUniqueIndexes(tx, obj, si, id)
if err != nil { if err != nil {
return fmt.Errorf("can't put unique indexes: %w", err) return fmt.Errorf("can't put unique indexes: %w", err)
} }