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

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2024-07-02 11:07:51 +03:00
parent 691af2b2eb
commit 06b0ad5b89

View file

@ -11,7 +11,6 @@ import (
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
"github.com/cockroachdb/pebble"
"github.com/dgraph-io/badger/v4"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
)
@ -68,61 +67,55 @@ func (db *DB) IterateCoveredByTombstones(ctx context.Context, tss map[string]oid
return ErrDegradedMode
}
return db.database.View(func(tx *badger.Txn) error {
return db.iterateCoveredByTombstones(ctx, tx, tss, h)
return db.snapshot(func(s *pebble.Snapshot) error {
return db.iterateCoveredByTombstones(ctx, s, tss, h)
})
}
func (db *DB) iterateCoveredByTombstones(ctx context.Context, tx *badger.Txn, tss map[string]oid.Address, h func(oid.Address) error) error {
func (db *DB) iterateCoveredByTombstones(ctx context.Context, r pebble.Reader, tss map[string]oid.Address, h func(oid.Address) error) error {
prefix := []byte{graveyardPrefix}
it := tx.NewIterator(badger.IteratorOptions{
PrefetchSize: badger.DefaultIteratorOptions.PrefetchSize,
Prefix: prefix,
PrefetchValues: true,
it, err := r.NewIter(&pebble.IterOptions{
LowerBound: prefix,
})
defer it.Close()
if err != nil {
return err
}
for it.Seek(nil); it.ValidForPrefix(prefix); it.Next() {
for v := it.First(); v && bytes.HasPrefix(it.Key(), prefix); v = it.Next() {
select {
case <-ctx.Done():
return ctx.Err()
return errors.Join(ctx.Err(), it.Close())
default:
}
var tombstoneAddress oid.Address
if err := it.Item().Value(func(val []byte) error {
var e error
tombstoneAddress, e = decodeAddressFromGrave(val)
return e
}); err != nil {
return err
tombstoneAddress, err := decodeAddressFromGrave(it.Value())
if err != nil {
return errors.Join(err, it.Close())
}
if _, ok := tss[tombstoneAddress.EncodeToString()]; !ok {
continue
}
var objectAddress oid.Address
var err error
objectAddress, err = addressFromGraveyardKey(it.Item().Key())
objectAddress, err := addressFromGraveyardKey(it.Key())
if err != nil {
return err
return errors.Join(err, it.Close())
}
isLocked, err := objectLocked(ctx, tx, objectAddress.Container(), objectAddress.Object())
isLocked, err := objectLocked(ctx, r, objectAddress.Container(), objectAddress.Object())
if err != nil {
return err
return errors.Join(err, it.Close())
}
if isLocked {
continue
}
if err := h(objectAddress); err != nil {
if errors.Is(err, ErrInterruptIterator) {
return nil
return it.Close()
}
return err
return errors.Join(err, it.Close())
}
}
return nil
return it.Close()
}
func iteratePhyObjects(r pebble.Reader, f func(cid.ID, oid.ID, *objectSDK.Object) error) error {