TrueCloudLab/frostfs-node
19
1
Fork 27
Code Issues 91 Pull requests 7 Projects Releases 1 Activity Actions

[#9999] metabase: Fix db engine to pebble in children.go

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2024-07-01 19:24:00 +03:00
parent c7955c63de
commit aa425c1b0c
1 changed files with 16 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
pkg/local_object_storage/metabase/children.go
View file

@ -7,7 +7,7 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr"
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"go.etcd.io/bbolt" "github.com/cockroachdb/pebble"
"go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace" "go.opentelemetry.io/otel/trace"
) )
@ -37,35 +37,31 @@ func (db *DB) GetChildren(ctx context.Context, addresses []oid.Address) (map[oid
} }
result := make(map[oid.Address][]oid.Address, len(addresses)) result := make(map[oid.Address][]oid.Address, len(addresses))
err := db.snapshot(func(s *pebble.Snapshot) error {
buffer := make([]byte, bucketKeySize)
err := db.database.View(func(tx *bbolt.Tx) error {
for _, addr := range addresses { for _, addr := range addresses {
if _, found := result[addr]; found { if _, found := result[addr]; found {
continue continue
} }
result[addr] = []oid.Address{} result[addr] = []oid.Address{}
bkt := tx.Bucket(parentBucketName(addr.Container(), buffer))
if bkt == nil {
continue
}
binObjIDs, err := decodeList(bkt.Get(objectKey(addr.Object(), buffer))) for {
keys, err := selectByPrefixBatch(ctx, s, parentKeyLongPrefix(addr.Container(), addr.Object()), batchSize)
if err != nil { if err != nil {
return err return err
} }
for _, binObjID := range binObjIDs { for _, key := range keys {
var id oid.ID resultAddress, err := addressOfTargetFromParentKey(key)
if err = id.Decode(binObjID); err != nil { if err != nil {
return err return err
} }
var resultAddress oid.Address
resultAddress.SetContainer(addr.Container())
resultAddress.SetObject(id)
result[addr] = append(result[addr], resultAddress) result[addr] = append(result[addr], resultAddress)
} }
if len(keys) < batchSize {
break
}
}
} }
return nil return nil
}) })