From f8e33f8e3ad90ff01d85c33cbeb21a5c9a7dfa16 Mon Sep 17 00:00:00 2001
From: Airat Arifullin <a.arifullin@yadro.com>
Date: Tue, 21 May 2024 14:25:15 +0300
Subject: [PATCH] [#1144] metabase: Proprely choose root OID for EC-splitted
 objects

* If EC-parent is a part of Split itself, then save to root bucket
  its parent;
* If EC-parent is not a part of Split itself, then save to root bucket
  OID of this EC-parent.

Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
---
 pkg/local_object_storage/metabase/put.go | 39 ++++++++++++++++++------
 1 file changed, 29 insertions(+), 10 deletions(-)

diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go
index bb5976ec..8a08adc3 100644
--- a/pkg/local_object_storage/metabase/put.go
+++ b/pkg/local_object_storage/metabase/put.go
@@ -257,17 +257,19 @@ func putUniqueIndexes(
 			}
 		}
 
-		err = putUniqueIndexItem(tx, namedBucketItem{
-			name: rootBucketName(cnr, bucketName),
-			key:  objKey,
-			val:  splitInfo,
-		})
-		if err != nil {
-			return err
-		}
-
+		isObjKeySet := true
 		if ecHead := obj.ECHeader(); ecHead != nil {
-			err = putECInfo(tx, cnr, objKey, ecHead)
+			if err = putECInfo(tx, cnr, objKey, ecHead); err != nil {
+				return err
+			}
+			objKey, isObjKeySet = objectKeyByECHeader(ecHead)
+		}
+		if isObjKeySet {
+			err = putUniqueIndexItem(tx, namedBucketItem{
+				name: rootBucketName(cnr, bucketName),
+				key:  objKey,
+				val:  splitInfo,
+			})
 			if err != nil {
 				return err
 			}
@@ -277,6 +279,23 @@ func putUniqueIndexes(
 	return nil
 }
 
+// objectKeyByECHeader returns objectKey for an object that has EC Header.
+// If object's parent is in Split, then parent's non-nil Split parent ID is set to object key.
+// If object's parent is not in Split, then its ID is set to object key.
+// Otherwise, such object keys should be ignored -- they are not put to the root bucket.
+func objectKeyByECHeader(ech *objectSDK.ECHeader) (objKey []byte, isSet bool) {
+	if ech.ParentSplitID() != nil {
+		if parentSplitParentID := ech.ParentSplitParentID(); parentSplitParentID != nil {
+			isSet = true
+			objKey = objectKey(*parentSplitParentID, make([]byte, objectKeySize))
+		}
+		return
+	}
+	isSet = true
+	objKey = objectKey(ech.Parent(), make([]byte, objectKeySize))
+	return
+}
+
 type updateIndexItemFunc = func(tx *bbolt.Tx, item namedBucketItem) error
 
 func updateListIndexes(tx *bbolt.Tx, obj *objectSDK.Object, f updateIndexItemFunc) error {