[#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>
This commit is contained in:
Airat Arifullin 2024-05-21 14:25:15 +03:00
parent f0edebea18
commit f8e33f8e3a

View file

@ -257,17 +257,19 @@ func putUniqueIndexes(
} }
} }
err = putUniqueIndexItem(tx, namedBucketItem{ isObjKeySet := true
name: rootBucketName(cnr, bucketName),
key: objKey,
val: splitInfo,
})
if err != nil {
return err
}
if ecHead := obj.ECHeader(); ecHead != nil { 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 { if err != nil {
return err return err
} }
@ -277,6 +279,23 @@ func putUniqueIndexes(
return nil 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 type updateIndexItemFunc = func(tx *bbolt.Tx, item namedBucketItem) error
func updateListIndexes(tx *bbolt.Tx, obj *objectSDK.Object, f updateIndexItemFunc) error { func updateListIndexes(tx *bbolt.Tx, obj *objectSDK.Object, f updateIndexItemFunc) error {