From 0e6fe4729247631a98adb634087bc1de90c284e0 Mon Sep 17 00:00:00 2001 From: Dmitrii Stepanov Date: Tue, 20 Feb 2024 16:44:23 +0300 Subject: [PATCH] [#977] metabase: Do not store root info for small objects Signed-off-by: Dmitrii Stepanov --- pkg/local_object_storage/metabase/put.go | 16 +++-------- pkg/local_object_storage/metabase/select.go | 30 ++++++++++++++++++++- 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go index 5d8e4d263..4f983a12a 100644 --- a/pkg/local_object_storage/metabase/put.go +++ b/pkg/local_object_storage/metabase/put.go @@ -248,19 +248,11 @@ func putUniqueIndexes( } // index root object - if obj.Type() == objectSDK.TypeRegular && !obj.HasParent() { - var ( - err error - splitInfo []byte - ) - - if isParent { - splitInfo, err = si.Marshal() - if err != nil { - return fmt.Errorf("can't marshal split info: %w", err) - } + if obj.Type() == objectSDK.TypeRegular && !obj.HasParent() && isParent { + splitInfo, err := si.Marshal() + if err != nil { + return fmt.Errorf("can't marshal split info: %w", err) } - err = putUniqueIndexItem(tx, namedBucketItem{ name: rootBucketName(cnr, bucketName), key: objKey, diff --git a/pkg/local_object_storage/metabase/select.go b/pkg/local_object_storage/metabase/select.go index 8b086a89f..fa527a062 100644 --- a/pkg/local_object_storage/metabase/select.go +++ b/pkg/local_object_storage/metabase/select.go @@ -181,6 +181,34 @@ func selectAllFromBucket(tx *bbolt.Tx, name []byte, to map[string]int, fNum int) }) } +// selectAllFromBucket goes through all keys in bucket and adds them in a +// resulting cache. Keys should be stringed object ids. +func selectRootObjects(tx *bbolt.Tx, cnr cid.ID, to map[string]int, fNum int) { + rootBkt := tx.Bucket(rootBucketName(cnr, make([]byte, bucketKeySize))) + if rootBkt != nil { + _ = rootBkt.ForEach(func(k, _ []byte) error { + markAddressInCache(to, fNum, string(k)) + return nil + }) + } + + primaryBkt := tx.Bucket(primaryBucketName(cnr, make([]byte, bucketKeySize))) + if primaryBkt == nil { + return + } + + _ = primaryBkt.ForEach(func(objectID, rawObject []byte) error { + obj := objectSDK.New() + if obj.Unmarshal(rawObject) != nil { + return nil + } + if !obj.HasParent() { + markAddressInCache(to, fNum, string(objectID)) + } + return nil + }) +} + // selectFastFilter makes fast optimized checks for well known buckets or // looking through user attribute buckets otherwise. func (db *DB) selectFastFilter( @@ -212,7 +240,7 @@ func (db *DB) selectFastFilter( bucketName := splitBucketName(cnr, bucketName) db.selectFromList(tx, bucketName, f, to, fNum) case v2object.FilterPropertyRoot: - selectAllFromBucket(tx, rootBucketName(cnr, bucketName), to, fNum) + selectRootObjects(tx, cnr, to, fNum) case v2object.FilterPropertyPhy: selectAllFromBucket(tx, primaryBucketName(cnr, bucketName), to, fNum) selectAllFromBucket(tx, tombstoneBucketName(cnr, bucketName), to, fNum)