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

[#308] object/relations: List relations in split order

Browse source 
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-11-17 16:30:47 +03:00
parent b4b07a3c4e
commit 9964a83083
2 changed files with 55 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
object/relations/relations.go
View file

@ -19,8 +19,9 @@ type Tokens struct {
}
type Relations interface {
// GetSplitInfo tries to get split info by root object id.
// If object isn't virtual it returns ErrNoSplitInfo.
// GetSplitInfo tries to get split info by some object id.
// This method must return split info on any object from split chain as well as on parent/linking object.
// If object doesn't have any split information returns ErrNoSplitInfo.
GetSplitInfo(ctx context.Context, cnrID cid.ID, rootID oid.ID, tokens Tokens) (*object.SplitInfo, error)
// ListChildrenByLinker returns list of children for link object.
@ -31,9 +32,6 @@ type Relations interface {
// If no previous object it returns ErrNoLeftSibling.
GetLeftSibling(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens Tokens) (oid.ID, error)
// FindSiblingBySplitID returns all objects that relates to the provided split id.
FindSiblingBySplitID(ctx context.Context, cnrID cid.ID, splitID *object.SplitID, tokens Tokens) ([]oid.ID, error)
// FindSiblingByParentID returns all object that relates to the provided parent id.
FindSiblingByParentID(ctx context.Context, cnrID cid.ID, parentID oid.ID, tokens Tokens) ([]oid.ID, error)
}
@ -46,9 +44,15 @@ var (
ErrNoSplitInfo = errors.New("no split info")
)
// ListAllRelations return all related phy objects for provided root object ID.
// Result doesn't include root object ID itself.
// ListAllRelations return all related phy objects for provided root object ID in split-chain order.
// Result doesn't include root object ID itself. If linking object is found its id will be the last one.
func ListAllRelations(ctx context.Context, rels Relations, cnrID cid.ID, rootObjID oid.ID, tokens Tokens) ([]oid.ID, error) {
return ListRelations(ctx, rels, cnrID, rootObjID, tokens, true)
}
// ListRelations return all related phy objects for provided root object ID in split-chain order.
// Result doesn't include root object ID itself.
func ListRelations(ctx context.Context, rels Relations, cnrID cid.ID, rootObjID oid.ID, tokens Tokens, includeLinking bool) ([]oid.ID, error) {
splitInfo, err := rels.GetSplitInfo(ctx, cnrID, rootObjID, tokens)
if err != nil {
if errors.Is(err, ErrNoSplitInfo) {
@ -59,22 +63,40 @@ func ListAllRelations(ctx context.Context, rels Relations, cnrID cid.ID, rootObj
// collect split chain by the descending ease of operations (ease is evaluated heuristically).
// If any approach fails, we don't try the next since we assume that it will fail too.
if _, ok := splitInfo.Link(); !ok {
// the list is expected to contain last part and (probably) split info
list, err := rels.FindSiblingByParentID(ctx, cnrID, rootObjID, tokens)
if err != nil {
return nil, fmt.Errorf("failed to find object children: %w", err)
}
for _, id := range list {
split, err := rels.GetSplitInfo(ctx, cnrID, id, tokens)
if err != nil {
if errors.Is(err, ErrNoSplitInfo) {
continue
}
return nil, fmt.Errorf("failed to get split info: %w", err)
}
if link, ok := split.Link(); ok {
splitInfo.SetLink(link)
}
if last, ok := split.LastPart(); ok {
splitInfo.SetLastPart(last)
}
}
}
if idLinking, ok := splitInfo.Link(); ok {
children, err := rels.ListChildrenByLinker(ctx, cnrID, idLinking, tokens)
if err != nil {
return nil, fmt.Errorf("failed to get linking object's header: %w", err)
}
// include linking object
return append(children, idLinking), nil
}
if idSplit := splitInfo.SplitID(); idSplit != nil {
members, err := rels.FindSiblingBySplitID(ctx, cnrID, idSplit, tokens)
if err != nil {
return nil, fmt.Errorf("failed to search objects by split ID: %w", err)
if includeLinking {
children = append(children, idLinking)
}
return members, nil
return children, nil
}
idMember, ok := splitInfo.LastPart()
@ -85,9 +107,6 @@ func ListAllRelations(ctx context.Context, rels Relations, cnrID cid.ID, rootObj
chain := []oid.ID{idMember}
chainSet := map[oid.ID]struct{}{idMember: {}}
// prmHead.SetRawFlag(false)
// split members are almost definitely singular, but don't get hung up on it
for {
idMember, err = rels.GetLeftSibling(ctx, cnrID, idMember, tokens)
if err != nil {
@ -101,20 +120,9 @@ func ListAllRelations(ctx context.Context, rels Relations, cnrID cid.ID, rootObj
return nil, fmt.Errorf("duplicated member in the split chain %s", idMember)
}
chain = append(chain, idMember)
chain = append([]oid.ID{idMember}, chain...)
chainSet[idMember] = struct{}{}
}
list, err := rels.FindSiblingByParentID(ctx, cnrID, rootObjID, tokens)
if err != nil {
return nil, fmt.Errorf("failed to find object children: %w", err)
}
for i := range list {
if _, ok = chainSet[list[i]]; !ok {
chain = append(chain, list[i])
}
}
return chain, nil
}

50
pool/pool.go
View file

@ -2510,7 +2510,7 @@ func (p *Pool) GetSplitInfo(ctx context.Context, cnrID cid.ID, objID oid.ID, tok
}
prm.MarkRaw()
_, err := p.HeadObject(ctx, prm)
res, err := p.HeadObject(ctx, prm)
var errSplit *object.SplitInfoError
@ -2518,7 +2518,21 @@ func (p *Pool) GetSplitInfo(ctx context.Context, cnrID cid.ID, objID oid.ID, tok
case errors.As(err, &errSplit):
return errSplit.SplitInfo(), nil
case err == nil:
return nil, relations.ErrNoSplitInfo
if res.SplitID() == nil {
return nil, relations.ErrNoSplitInfo
}
splitInfo := object.NewSplitInfo()
splitInfo.SetSplitID(res.SplitID())
if res.HasParent() {
if len(res.Children()) > 0 {
splitInfo.SetLink(objID)
} else {
splitInfo.SetLastPart(objID)
}
}
return splitInfo, nil
default:
return nil, fmt.Errorf("failed to get raw object header: %w", err)
}
@ -2574,38 +2588,6 @@ func (p *Pool) GetLeftSibling(ctx context.Context, cnrID cid.ID, objID oid.ID, t
return idMember, nil
}
// FindSiblingBySplitID implements relations.Relations.
func (p *Pool) FindSiblingBySplitID(ctx context.Context, cnrID cid.ID, splitID *object.SplitID, tokens relations.Tokens) ([]oid.ID, error) {
var query object.SearchFilters
query.AddSplitIDFilter(object.MatchStringEqual, splitID)
var prm PrmObjectSearch
prm.SetContainerID(cnrID)
prm.SetFilters(query)
if tokens.Bearer != nil {
prm.UseBearer(*tokens.Bearer)
}
if tokens.Session != nil {
prm.UseSession(*tokens.Session)
}
res, err := p.SearchObjects(ctx, prm)
if err != nil {
return nil, fmt.Errorf("failed to search objects by split ID: %w", err)
}
var members []oid.ID
err = res.Iterate(func(id oid.ID) bool {
members = append(members, id)
return false
})
if err != nil {
return nil, fmt.Errorf("failed to iterate found objects: %w", err)
}
return members, nil
}
// FindSiblingByParentID implements relations.Relations.
func (p *Pool) FindSiblingByParentID(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) ([]oid.ID, error) {
var query object.SearchFilters