satellite/metabase: return more information from delete last committed

Change-Id: I2626a100e0c3c41631c9a29b0bf5a7afccc60957
This commit is contained in:
Egon Elbre 2023-10-23 17:06:01 +03:00
parent 97c98d72e4
commit a7e1378f89
8 changed files with 168 additions and 98 deletions

View File

@ -71,7 +71,9 @@ func TestCollectBucketTallies(t *testing.T) {
ObjectKey: randStream.ObjectKey, ObjectKey: randStream.ObjectKey,
}, },
}, },
Result: metabase.DeleteObjectResult{Objects: []metabase.Object{obj}}, Result: metabase.DeleteObjectResult{
Removed: []metabase.Object{obj},
},
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.CollectBucketTallies{ metabasetest.CollectBucketTallies{

View File

@ -45,7 +45,10 @@ func (obj *DeleteObjectExactVersion) Verify() error {
// DeleteObjectResult result of deleting object. // DeleteObjectResult result of deleting object.
type DeleteObjectResult struct { type DeleteObjectResult struct {
Objects []Object // Removed contains the list of objects that were removed from the metabase.
Removed []Object
// Markers contains the delete markers that were added.
Markers []Object
} }
// DeleteObjectsAllVersions contains arguments necessary for deleting all versions of multiple objects from the same bucket. // DeleteObjectsAllVersions contains arguments necessary for deleting all versions of multiple objects from the same bucket.
@ -129,15 +132,15 @@ func (db *DB) deleteObjectExactVersion(ctx context.Context, opts DeleteObjectExa
FROM deleted_objects`, FROM deleted_objects`,
opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version), opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version),
)(func(rows tagsql.Rows) error { )(func(rows tagsql.Rows) error {
result.Objects, err = db.scanObjectDeletion(ctx, opts.ObjectLocation, rows) result.Removed, err = db.scanObjectDeletion(ctx, opts.ObjectLocation, rows)
return err return err
}) })
if err != nil { if err != nil {
return DeleteObjectResult{}, err return DeleteObjectResult{}, err
} }
mon.Meter("object_delete").Mark(len(result.Objects)) mon.Meter("object_delete").Mark(len(result.Removed))
for _, object := range result.Objects { for _, object := range result.Removed {
mon.Meter("segment_delete").Mark(int(object.SegmentCount)) mon.Meter("segment_delete").Mark(int(object.SegmentCount))
} }
@ -186,7 +189,7 @@ func (db *DB) DeletePendingObject(ctx context.Context, opts DeletePendingObject)
total_plain_size, total_encrypted_size, fixed_segment_size, encryption total_plain_size, total_encrypted_size, fixed_segment_size, encryption
FROM deleted_objects FROM deleted_objects
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version, opts.StreamID))(func(rows tagsql.Rows) error { `, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version, opts.StreamID))(func(rows tagsql.Rows) error {
result.Objects, err = db.scanObjectDeletion(ctx, opts.Location(), rows) result.Removed, err = db.scanObjectDeletion(ctx, opts.Location(), rows)
return err return err
}) })
@ -194,12 +197,12 @@ func (db *DB) DeletePendingObject(ctx context.Context, opts DeletePendingObject)
return DeleteObjectResult{}, err return DeleteObjectResult{}, err
} }
if len(result.Objects) == 0 { if len(result.Removed) == 0 {
return DeleteObjectResult{}, ErrObjectNotFound.Wrap(Error.New("no rows deleted")) return DeleteObjectResult{}, ErrObjectNotFound.Wrap(Error.New("no rows deleted"))
} }
mon.Meter("object_delete").Mark(len(result.Objects)) mon.Meter("object_delete").Mark(len(result.Removed))
for _, object := range result.Objects { for _, object := range result.Removed {
mon.Meter("segment_delete").Mark(int(object.SegmentCount)) mon.Meter("segment_delete").Mark(int(object.SegmentCount))
} }
@ -231,23 +234,24 @@ func (db *DB) DeletePendingObjectNew(ctx context.Context, opts DeletePendingObje
) )
SELECT * FROM deleted_objects SELECT * FROM deleted_objects
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.StreamID))(func(rows tagsql.Rows) error { `, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.StreamID))(func(rows tagsql.Rows) error {
result.Objects, err = db.scanPendingObjectDeletion(ctx, opts.Location(), rows) result.Removed, err = db.scanPendingObjectDeletion(ctx, opts.Location(), rows)
return err return err
}) })
if err != nil { if err != nil {
return DeleteObjectResult{}, err return DeleteObjectResult{}, err
} }
if len(result.Objects) == 0 { if len(result.Removed) == 0 {
return DeleteObjectResult{}, ErrObjectNotFound.Wrap(Error.New("no rows deleted")) return DeleteObjectResult{}, ErrObjectNotFound.Wrap(Error.New("no rows deleted"))
} }
mon.Meter("object_delete").Mark(len(result.Objects)) mon.Meter("object_delete").Mark(len(result.Removed))
return result, nil return result, nil
} }
type deleteObjectUnversionedCommittedResult struct { type deleteObjectUnversionedCommittedResult struct {
Deleted []Object
// DeletedObjectCount and DeletedSegmentCount return how many elements were deleted. // DeletedObjectCount and DeletedSegmentCount return how many elements were deleted.
DeletedObjectCount int DeletedObjectCount int
DeletedSegmentCount int DeletedSegmentCount int
@ -266,6 +270,15 @@ func (db *DB) deleteObjectUnversionedCommitted(ctx context.Context, loc ObjectLo
return deleteObjectUnversionedCommittedResult{}, Error.Wrap(err) return deleteObjectUnversionedCommittedResult{}, Error.Wrap(err)
} }
var deleted Object
var version sql.NullInt64
var streamID uuid.NullUUID
var createdAt sql.NullTime
var status sql.NullByte
var params nullableValue[encryptionParameters]
params.value.EncryptionParameters = &deleted.Encryption
err = stmt.QueryRowContext(ctx, ` err = stmt.QueryRowContext(ctx, `
WITH highest_object AS ( WITH highest_object AS (
SELECT MAX(version) as version SELECT MAX(version) as version
@ -276,18 +289,53 @@ func (db *DB) deleteObjectUnversionedCommitted(ctx context.Context, loc ObjectLo
WHERE WHERE
(project_id, bucket_name, object_key) = ($1, $2, $3) (project_id, bucket_name, object_key) = ($1, $2, $3)
AND status IN `+statusesUnversioned+` AND status IN `+statusesUnversioned+`
RETURNING stream_id RETURNING
version, stream_id, status, created_at, expires_at,
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
encryption
), deleted_segments AS ( ), deleted_segments AS (
DELETE FROM segments DELETE FROM segments
WHERE segments.stream_id IN (SELECT deleted_objects.stream_id FROM deleted_objects) WHERE segments.stream_id IN (SELECT deleted_objects.stream_id FROM deleted_objects)
RETURNING segments.stream_id RETURNING segments.stream_id
) )
SELECT SELECT
(SELECT version FROM deleted_objects),
(SELECT stream_id FROM deleted_objects),
(SELECT status FROM deleted_objects),
(SELECT created_at FROM deleted_objects),
(SELECT expires_at FROM deleted_objects),
(SELECT encrypted_metadata_nonce FROM deleted_objects),
(SELECT encrypted_metadata FROM deleted_objects),
(SELECT encrypted_metadata_encrypted_key FROM deleted_objects),
(SELECT encryption FROM deleted_objects),
(SELECT count(*) FROM deleted_objects), (SELECT count(*) FROM deleted_objects),
(SELECT count(*) FROM deleted_segments), (SELECT count(*) FROM deleted_segments),
coalesce((SELECT version FROM highest_object), 0) coalesce((SELECT version FROM highest_object), 0)
`, loc.ProjectID, []byte(loc.BucketName), loc.ObjectKey). `, loc.ProjectID, []byte(loc.BucketName), loc.ObjectKey).
Scan(&result.DeletedObjectCount, &result.DeletedSegmentCount, &result.MaxVersion) Scan(
&version,
&streamID,
&status,
&createdAt,
&deleted.ExpiresAt,
&deleted.EncryptedMetadataNonce,
&deleted.EncryptedMetadata,
&deleted.EncryptedMetadataEncryptedKey,
&params,
&result.DeletedObjectCount,
&result.DeletedSegmentCount,
&result.MaxVersion,
)
deleted.ProjectID = loc.ProjectID
deleted.BucketName = loc.BucketName
deleted.ObjectKey = loc.ObjectKey
deleted.Version = Version(version.Int64)
deleted.Status = ObjectStatus(status.Byte)
deleted.StreamID = streamID.UUID
deleted.CreatedAt = createdAt.Time
if err != nil { if err != nil {
return deleteObjectUnversionedCommittedResult{}, Error.Wrap(err) return deleteObjectUnversionedCommittedResult{}, Error.Wrap(err)
@ -307,6 +355,10 @@ func (db *DB) deleteObjectUnversionedCommitted(ctx context.Context, loc ObjectLo
return result, Error.New("internal error: multiple committed unversioned objects") return result, Error.New("internal error: multiple committed unversioned objects")
} }
if result.DeletedObjectCount > 0 {
result.Deleted = append(result.Deleted, deleted)
}
return result, nil return result, nil
} }
@ -389,7 +441,7 @@ func (db *DB) DeleteObjectsAllVersions(ctx context.Context, opts DeleteObjectsAl
fixed_segment_size, encryption fixed_segment_size, encryption
FROM deleted_objects FROM deleted_objects
`, projectID, []byte(bucketName), pgutil.ByteaArray(objectKeys)))(func(rows tagsql.Rows) error { `, projectID, []byte(bucketName), pgutil.ByteaArray(objectKeys)))(func(rows tagsql.Rows) error {
result.Objects, err = db.scanMultipleObjectsDeletion(ctx, rows) result.Removed, err = db.scanMultipleObjectsDeletion(ctx, rows)
return err return err
}) })
@ -397,8 +449,8 @@ func (db *DB) DeleteObjectsAllVersions(ctx context.Context, opts DeleteObjectsAl
return DeleteObjectResult{}, err return DeleteObjectResult{}, err
} }
mon.Meter("object_delete").Mark(len(result.Objects)) mon.Meter("object_delete").Mark(len(result.Removed))
for _, object := range result.Objects { for _, object := range result.Removed {
mon.Meter("segment_delete").Mark(int(object.SegmentCount)) mon.Meter("segment_delete").Mark(int(object.SegmentCount))
} }
@ -518,7 +570,6 @@ func (db *DB) DeleteObjectLastCommitted(
err = txutil.WithTx(ctx, db.db, nil, func(ctx context.Context, tx tagsql.Tx) (err error) { err = txutil.WithTx(ctx, db.db, nil, func(ctx context.Context, tx tagsql.Tx) (err error) {
// TODO(ver) fold deleteObjectUnversionedCommitted into query below using ON CONFLICT // TODO(ver) fold deleteObjectUnversionedCommitted into query below using ON CONFLICT
deleted, err := db.deleteObjectUnversionedCommitted(ctx, opts.ObjectLocation, tx) deleted, err := db.deleteObjectUnversionedCommitted(ctx, opts.ObjectLocation, tx)
// TODO(ver): should we return in the result as well?
if err != nil { if err != nil {
return Error.Wrap(err) return Error.Wrap(err)
} }
@ -552,7 +603,8 @@ func (db *DB) DeleteObjectLastCommitted(
return Error.Wrap(err) return Error.Wrap(err)
} }
result.Objects = append(result.Objects, marker) result.Markers = append(result.Markers, marker)
result.Removed = deleted.Deleted
return nil return nil
}) })
return result, err return result, err
@ -594,7 +646,6 @@ func (db *DB) DeleteObjectLastCommitted(
SELECT version, created_at FROM added_object SELECT version, created_at FROM added_object
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, streamID) `, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, streamID)
// TODO(ver): should this return the deleted object or the delete marker?
var deleted Object var deleted Object
deleted.ProjectID = opts.ProjectID deleted.ProjectID = opts.ProjectID
deleted.BucketName = opts.BucketName deleted.BucketName = opts.BucketName
@ -609,7 +660,7 @@ func (db *DB) DeleteObjectLastCommitted(
} }
return DeleteObjectResult{}, Error.Wrap(err) return DeleteObjectResult{}, Error.Wrap(err)
} }
return DeleteObjectResult{Objects: []Object{deleted}}, nil return DeleteObjectResult{Markers: []Object{deleted}}, nil
} }
// TODO(ver): do we need to pretend here that `expires_at` matters? // TODO(ver): do we need to pretend here that `expires_at` matters?
@ -641,15 +692,15 @@ func (db *DB) DeleteObjectLastCommitted(
FROM deleted_objects`, FROM deleted_objects`,
opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey), opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey),
)(func(rows tagsql.Rows) error { )(func(rows tagsql.Rows) error {
result.Objects, err = db.scanObjectDeletion(ctx, opts.ObjectLocation, rows) result.Removed, err = db.scanObjectDeletion(ctx, opts.ObjectLocation, rows)
return err return err
}) })
if err != nil { if err != nil {
return DeleteObjectResult{}, err return DeleteObjectResult{}, err
} }
mon.Meter("object_delete").Mark(len(result.Objects)) mon.Meter("object_delete").Mark(len(result.Removed))
for _, object := range result.Objects { for _, object := range result.Removed {
mon.Meter("segment_delete").Mark(int(object.SegmentCount)) mon.Meter("segment_delete").Mark(int(object.SegmentCount))
} }

View File

@ -182,7 +182,7 @@ func TestDeletePendingObject(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{metabase.Object(object)}, Removed: []metabase.Object{metabase.Object(object)},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -199,7 +199,7 @@ func TestDeletePendingObject(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Removed: []metabase.Object{
{ {
ObjectStream: obj, ObjectStream: obj,
CreatedAt: now, CreatedAt: now,
@ -243,7 +243,7 @@ func TestDeletePendingObject(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Removed: []metabase.Object{
{ {
ObjectStream: obj, ObjectStream: obj,
CreatedAt: now, CreatedAt: now,
@ -420,7 +420,7 @@ func TestDeletePendingObjectNew(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{metabase.Object(object)}, Removed: []metabase.Object{metabase.Object(object)},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -467,7 +467,7 @@ func TestDeletePendingObjectNew(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Removed: []metabase.Object{
{ {
ObjectStream: obj, ObjectStream: obj,
CreatedAt: now, CreatedAt: now,
@ -516,7 +516,7 @@ func TestDeletePendingObjectNew(t *testing.T) {
ObjectStream: obj, ObjectStream: obj,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Removed: []metabase.Object{
{ {
ObjectStream: obj, ObjectStream: obj,
CreatedAt: now, CreatedAt: now,
@ -578,7 +578,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: 1, Version: 1,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{}, Removed: []metabase.Object{},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{}.Check(ctx, t, db) metabasetest.Verify{}.Check(ctx, t, db)
@ -593,7 +593,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: 33, Version: 33,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{}, Removed: []metabase.Object{},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{}.Check(ctx, t, db) metabasetest.Verify{}.Check(ctx, t, db)
@ -615,7 +615,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: obj.Version, Version: obj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{{ Removed: []metabase.Object{{
ObjectStream: obj, ObjectStream: obj,
CreatedAt: now, CreatedAt: now,
Encryption: metabasetest.DefaultEncryption, Encryption: metabasetest.DefaultEncryption,
@ -649,7 +649,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: obj.Version, Version: obj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -667,7 +667,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: obj.Version, Version: obj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -711,7 +711,7 @@ func TestDeleteObjectExactVersion(t *testing.T) {
Version: obj.Version, Version: obj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -792,7 +792,7 @@ func TestDeleteObjectVersioning(t *testing.T) {
Versioned: true, Versioned: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Markers: []metabase.Object{
{ {
ObjectStream: marker, ObjectStream: marker,
CreatedAt: now, CreatedAt: now,
@ -839,7 +839,7 @@ func TestDeleteObjectVersioning(t *testing.T) {
Versioned: true, Versioned: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Markers: []metabase.Object{
{ {
ObjectStream: marker, ObjectStream: marker,
CreatedAt: now, CreatedAt: now,
@ -857,7 +857,7 @@ func TestDeleteObjectVersioning(t *testing.T) {
Versioned: true, Versioned: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Markers: []metabase.Object{
{ {
ObjectStream: marker2, ObjectStream: marker2,
CreatedAt: now, CreatedAt: now,
@ -1006,7 +1006,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Locations: []metabase.ObjectLocation{location}, Locations: []metabase.ObjectLocation{location},
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1023,7 +1023,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Locations: []metabase.ObjectLocation{location}, Locations: []metabase.ObjectLocation{location},
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1045,7 +1045,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Locations: []metabase.ObjectLocation{location, obj2.Location()}, Locations: []metabase.ObjectLocation{location, obj2.Location()},
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object1, object2}, Removed: []metabase.Object{object1, object2},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1088,7 +1088,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Locations: []metabase.ObjectLocation{location}, Locations: []metabase.ObjectLocation{location},
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1137,7 +1137,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Locations: []metabase.ObjectLocation{location, object2.Location()}, Locations: []metabase.ObjectLocation{location, object2.Location()},
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object1, object2}, Removed: []metabase.Object{object1, object2},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1152,7 +1152,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
for i := 1; i <= 10; i++ { for i := 1; i <= 10; i++ {
obj.StreamID = testrand.UUID() obj.StreamID = testrand.UUID()
obj.Version = metabase.Version(i) obj.Version = metabase.Version(i)
expected.Objects = append(expected.Objects, metabasetest.CreateObjectVersioned(ctx, t, db, obj, 1)) expected.Removed = append(expected.Removed, metabasetest.CreateObjectVersioned(ctx, t, db, obj, 1))
} }
metabasetest.DeleteObjectsAllVersions{ metabasetest.DeleteObjectsAllVersions{
@ -1171,7 +1171,7 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
now := time.Now() now := time.Now()
obj := metabasetest.RandObjectStream() obj := metabasetest.RandObjectStream()
_ = metabasetest.CreateObject(ctx, t, db, obj, 0) object := metabasetest.CreateObject(ctx, t, db, obj, 0)
marker := metabase.Object{ marker := metabase.Object{
ObjectStream: metabase.ObjectStream{ ObjectStream: metabase.ObjectStream{
@ -1194,7 +1194,12 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Versioned: false, Versioned: false,
Suspended: true, Suspended: true,
}, },
Result: metabase.DeleteObjectResult{Objects: []metabase.Object{marker}}, Result: metabase.DeleteObjectResult{
Markers: []metabase.Object{marker},
Removed: []metabase.Object{
object,
},
},
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{ metabasetest.Verify{
@ -1233,7 +1238,9 @@ func TestDeleteObjectsAllVersions(t *testing.T) {
Versioned: false, Versioned: false,
Suspended: true, Suspended: true,
}, },
Result: metabase.DeleteObjectResult{Objects: []metabase.Object{marker}}, Result: metabase.DeleteObjectResult{
Markers: []metabase.Object{marker},
},
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{ metabasetest.Verify{
@ -1282,7 +1289,7 @@ func TestDeleteCopyWithDuplicateMetadata(t *testing.T) {
Version: copyObj.Version, Version: copyObj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{copyObj}, Removed: []metabase.Object{copyObj},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1321,7 +1328,7 @@ func TestDeleteCopyWithDuplicateMetadata(t *testing.T) {
Version: copyObject1.Version, Version: copyObject1.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{copyObject1}, Removed: []metabase.Object{copyObject1},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1358,7 +1365,7 @@ func TestDeleteCopyWithDuplicateMetadata(t *testing.T) {
Version: originalObj.Version, Version: originalObj.Version,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{originalObj}, Removed: []metabase.Object{originalObj},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1471,9 +1478,7 @@ func TestDeleteObjectLastCommitted(t *testing.T) {
Opts: metabase.DeleteObjectLastCommitted{ Opts: metabase.DeleteObjectLastCommitted{
ObjectLocation: location, ObjectLocation: location,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{},
Objects: []metabase.Object{},
},
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{}.Check(ctx, t, db) metabasetest.Verify{}.Check(ctx, t, db)
}) })
@ -1498,7 +1503,7 @@ func TestDeleteObjectLastCommitted(t *testing.T) {
ObjectLocation: location, ObjectLocation: location,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1515,7 +1520,7 @@ func TestDeleteObjectLastCommitted(t *testing.T) {
ObjectLocation: location, ObjectLocation: location,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1555,7 +1560,7 @@ func TestDeleteObjectLastCommitted(t *testing.T) {
ObjectLocation: location, ObjectLocation: location,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{object}, Removed: []metabase.Object{object},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -1610,7 +1615,9 @@ func TestDeleteObjectLastCommitted(t *testing.T) {
BucketName: newObj.BucketName, BucketName: newObj.BucketName,
ObjectKey: newObj.ObjectKey, ObjectKey: newObj.ObjectKey,
}}, }},
Result: metabase.DeleteObjectResult{Objects: []metabase.Object{committedObject}}, Result: metabase.DeleteObjectResult{
Removed: []metabase.Object{committedObject},
},
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{ metabasetest.Verify{

View File

@ -4,6 +4,7 @@
package metabase package metabase
import ( import (
"database/sql"
"database/sql/driver" "database/sql/driver"
"encoding/binary" "encoding/binary"
@ -12,6 +13,20 @@ import (
"storj.io/common/storj" "storj.io/common/storj"
) )
type nullableValue[T sql.Scanner] struct {
isnull bool
value T
}
func (v *nullableValue[T]) Scan(value interface{}) error {
if value == nil {
v.isnull = true
return nil
}
v.isnull = false
return v.value.Scan(value)
}
type encryptionParameters struct { type encryptionParameters struct {
*storj.EncryptionParameters *storj.EncryptionParameters
} }

View File

@ -197,7 +197,7 @@ func TestGetObjectExactVersion(t *testing.T) {
Versioned: true, Versioned: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{versionedMarker}, Markers: []metabase.Object{versionedMarker},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -225,7 +225,8 @@ func TestGetObjectExactVersion(t *testing.T) {
Suspended: true, Suspended: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{unversionedMarker}, Removed: []metabase.Object{unversioned},
Markers: []metabase.Object{unversionedMarker},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -438,7 +439,7 @@ func TestGetObjectLastCommitted(t *testing.T) {
}.Check(ctx, t, db) }.Check(ctx, t, db)
metabasetest.Verify{Objects: []metabase.RawObject{ metabasetest.Verify{Objects: []metabase.RawObject{
metabase.RawObject(result.Objects[0]), metabase.RawObject(result.Markers[0]),
metabase.RawObject(firstObject), metabase.RawObject(firstObject),
metabase.RawObject(secondObject), metabase.RawObject(secondObject),
}}.Check(ctx, t, db) }}.Check(ctx, t, db)
@ -461,7 +462,7 @@ func TestGetObjectLastCommitted(t *testing.T) {
ObjectLocation: obj.Location(), ObjectLocation: obj.Location(),
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{originalObject}, Removed: []metabase.Object{originalObject},
}, },
}.Check(ctx, t, db) }.Check(ctx, t, db)

View File

@ -966,7 +966,7 @@ func TestListObjectsVersioned(t *testing.T) {
Versioned: true, Versioned: true,
}, },
Result: metabase.DeleteObjectResult{ Result: metabase.DeleteObjectResult{
Objects: []metabase.Object{ Markers: []metabase.Object{
{ {
ObjectStream: metabase.ObjectStream{ ObjectStream: metabase.ObjectStream{
ProjectID: objA0.ProjectID, ProjectID: objA0.ProjectID,
@ -998,7 +998,7 @@ func TestListObjectsVersioned(t *testing.T) {
metabasetest.Verify{ metabasetest.Verify{
Objects: []metabase.RawObject{ Objects: []metabase.RawObject{
metabase.RawObject(deletionResult.Objects[0]), metabase.RawObject(deletionResult.Markers[0]),
metabase.RawObject(objA0), metabase.RawObject(objA0),
metabase.RawObject(objA1), metabase.RawObject(objA1),
metabase.RawObject(objB0), metabase.RawObject(objB0),

View File

@ -416,16 +416,24 @@ type DeleteObjectExactVersion struct {
ErrText string ErrText string
} }
func compareDeleteObjectResult(t testing.TB, got, exp metabase.DeleteObjectResult) {
t.Helper()
sortObjects(got.Markers)
sortObjects(exp.Markers)
sortObjects(got.Removed)
sortObjects(exp.Removed)
diff := cmp.Diff(exp, got, DefaultTimeDiff(), cmpopts.EquateEmpty())
require.Zero(t, diff)
}
// Check runs the test. // Check runs the test.
func (step DeleteObjectExactVersion) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) { func (step DeleteObjectExactVersion) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) {
result, err := db.DeleteObjectExactVersion(ctx, step.Opts) result, err := db.DeleteObjectExactVersion(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText) checkError(t, err, step.ErrClass, step.ErrText)
compareDeleteObjectResult(t, result, step.Result)
sortObjects(result.Objects)
sortObjects(step.Result.Objects)
diff := cmp.Diff(step.Result, result, DefaultTimeDiff(), cmpopts.EquateEmpty())
require.Zero(t, diff)
} }
// DeletePendingObject is for testing metabase.DeletePendingObject. // DeletePendingObject is for testing metabase.DeletePendingObject.
@ -440,12 +448,7 @@ type DeletePendingObject struct {
func (step DeletePendingObject) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) { func (step DeletePendingObject) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) {
result, err := db.DeletePendingObject(ctx, step.Opts) result, err := db.DeletePendingObject(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText) checkError(t, err, step.ErrClass, step.ErrText)
compareDeleteObjectResult(t, result, step.Result)
sortObjects(result.Objects)
sortObjects(step.Result.Objects)
diff := cmp.Diff(step.Result, result, DefaultTimeDiff())
require.Zero(t, diff)
} }
// DeletePendingObjectNew is for testing metabase.DeletePendingObjectNew. // DeletePendingObjectNew is for testing metabase.DeletePendingObjectNew.
@ -460,12 +463,7 @@ type DeletePendingObjectNew struct {
func (step DeletePendingObjectNew) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) { func (step DeletePendingObjectNew) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) {
result, err := db.DeletePendingObjectNew(ctx, step.Opts) result, err := db.DeletePendingObjectNew(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText) checkError(t, err, step.ErrClass, step.ErrText)
compareDeleteObjectResult(t, result, step.Result)
sortObjects(result.Objects)
sortObjects(step.Result.Objects)
diff := cmp.Diff(step.Result, result, DefaultTimeDiff())
require.Zero(t, diff)
} }
// DeleteObjectsAllVersions is for testing metabase.DeleteObjectsAllVersions. // DeleteObjectsAllVersions is for testing metabase.DeleteObjectsAllVersions.
@ -480,12 +478,7 @@ type DeleteObjectsAllVersions struct {
func (step DeleteObjectsAllVersions) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) { func (step DeleteObjectsAllVersions) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) {
result, err := db.DeleteObjectsAllVersions(ctx, step.Opts) result, err := db.DeleteObjectsAllVersions(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText) checkError(t, err, step.ErrClass, step.ErrText)
compareDeleteObjectResult(t, result, step.Result)
sortObjects(result.Objects)
sortObjects(step.Result.Objects)
diff := cmp.Diff(step.Result, result, DefaultTimeDiff())
require.Zero(t, diff)
} }
// DeleteExpiredObjects is for testing metabase.DeleteExpiredObjects. // DeleteExpiredObjects is for testing metabase.DeleteExpiredObjects.
@ -796,13 +789,7 @@ type DeleteObjectLastCommitted struct {
func (step DeleteObjectLastCommitted) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) metabase.DeleteObjectResult { func (step DeleteObjectLastCommitted) Check(ctx *testcontext.Context, t testing.TB, db *metabase.DB) metabase.DeleteObjectResult {
result, err := db.DeleteObjectLastCommitted(ctx, step.Opts) result, err := db.DeleteObjectLastCommitted(ctx, step.Opts)
checkError(t, err, step.ErrClass, step.ErrText) checkError(t, err, step.ErrClass, step.ErrText)
compareDeleteObjectResult(t, result, step.Result)
sortObjects(result.Objects)
sortObjects(step.Result.Objects)
diff := cmp.Diff(step.Result, result, DefaultTimeDiff(), cmpopts.EquateEmpty())
require.Zero(t, diff)
return result return result
} }

View File

@ -1819,13 +1819,20 @@ func (endpoint *Endpoint) DeletePendingObject(ctx context.Context, stream metaba
} }
func (endpoint *Endpoint) deleteObjectResultToProto(ctx context.Context, result metabase.DeleteObjectResult) (deletedObjects []*pb.Object, err error) { func (endpoint *Endpoint) deleteObjectResultToProto(ctx context.Context, result metabase.DeleteObjectResult) (deletedObjects []*pb.Object, err error) {
deletedObjects = make([]*pb.Object, len(result.Objects)) deletedObjects = make([]*pb.Object, 0, len(result.Removed)+len(result.Markers))
for i, object := range result.Objects { for _, object := range result.Removed {
deletedObject, err := endpoint.objectToProto(ctx, object, endpoint.defaultRS) deletedObject, err := endpoint.objectToProto(ctx, object, endpoint.defaultRS)
if err != nil { if err != nil {
return nil, err return nil, err
} }
deletedObjects[i] = deletedObject deletedObjects = append(deletedObjects, deletedObject)
}
for _, object := range result.Markers {
deletedObject, err := endpoint.objectToProto(ctx, object, endpoint.defaultRS)
if err != nil {
return nil, err
}
deletedObjects = append(deletedObjects, deletedObject)
} }
return deletedObjects, nil return deletedObjects, nil