From 0e6fe4729247631a98adb634087bc1de90c284e0 Mon Sep 17 00:00:00 2001
From: Dmitrii Stepanov <d.stepanov@yadro.com>
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 <d.stepanov@yadro.com>
---
 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)
-- 
2.45.2