TrueCloudLab/frostfs-node
21
1
Fork 27
Code Issues 108 Pull requests 3 Projects Releases 2 Activity Actions

[#158] metabase: Implement operation of deleting a group of objects

Browse source 
Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2020-11-09 13:09:09 +03:00 committed by Alex Vanin
parent 0021f1cc5f
commit 0cd05fdca5
2 changed files with 40 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
pkg/local_object_storage/metabase/delete.go
View file

@ -10,18 +10,7 @@ var tombstoneBucket = []byte("tombstones")
// Delete marks object as deleted.
func (db *DB) Delete(addr *object.Address) error {
return db.boltDB.Update(func(tx *bbolt.Tx) error {
bucket, err := tx.CreateBucketIfNotExists(tombstoneBucket)
if err != nil {
return errors.Wrapf(err, "(%T) could not create tombstone bucket", db)
}
if err := bucket.Put(addressKey(addr), nil); err != nil {
return errors.Wrapf(err, "(%T) could not put to tombstone bucket", db)
}
return nil
})
return db.delete(addr)
}
func objectRemoved(tx *bbolt.Tx, addr []byte) bool {
@ -29,3 +18,27 @@ func objectRemoved(tx *bbolt.Tx, addr []byte) bool {
return tombstoneBucket != nil && tombstoneBucket.Get(addr) != nil
}
// DeleteObjects marks list of objects as deleted.
func (db *DB) DeleteObjects(list ...*object.Address) {
if err := db.delete(list...); err != nil {
// TODO: log error
}
}
func (db *DB) delete(list ...*object.Address) error {
return db.boltDB.Update(func(tx *bbolt.Tx) error {
bucket, err := tx.CreateBucketIfNotExists(tombstoneBucket)
if err != nil {
return errors.Wrapf(err, "(%T) could not create tombstone bucket", db)
}
for i := range list {
if err := bucket.Put(addressKey(list[i]), nil); err != nil {
return errors.Wrapf(err, "(%T) could not put to tombstone bucket", db)
}
}
return nil
})
}

15
pkg/local_object_storage/metabase/delete_test.go
View file

@ -54,3 +54,18 @@ func BenchmarkDB_Delete(b *testing.B) {
}
})
}
func TestDB_DeleteObjects(t *testing.T) {
db := newDB(t)
defer releaseDB(db)
o1 := generateObject(t, testPrm{})
o2 := generateObject(t, testPrm{})
require.NoError(t, db.Put(o1))
require.NoError(t, db.Put(o2))
db.DeleteObjects(o1.Address(), o2.Address())
testSelect(t, db, object.SearchFilters{})
}