JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/metabase: adjust code for iteration

Browse Source 
Change-Id: Id3d4efe228a6f2d3642a639ef66a30e178ca001a
This commit is contained in:
Egon Elbre 2023-10-18 17:51:27 +03:00 committed by Storj Robot
parent 988ebbaf8d
commit 97c98d72e4
6 changed files with 86 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
satellite/metabase/bench_test.go
View File

@ -217,7 +217,7 @@ func (s *scenario) run(ctx *testcontext.Context, b *testing.B, db *metabase.DB)
err := db.IterateObjectsAllVersionsWithStatus(ctx, metabase.IterateObjectsWithStatus{ err := db.IterateObjectsAllVersionsWithStatus(ctx, metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: "bucket", BucketName: "bucket",
Status: metabase.CommittedUnversioned, Pending: false,
}, func(ctx context.Context, it metabase.ObjectsIterator) error { }, func(ctx context.Context, it metabase.ObjectsIterator) error {
var entry metabase.ObjectEntry var entry metabase.ObjectEntry
for it.Next(ctx, &entry) { for it.Next(ctx, &entry) {
@ -241,7 +241,7 @@ func (s *scenario) run(ctx *testcontext.Context, b *testing.B, db *metabase.DB)
ProjectID: projectID, ProjectID: projectID,
BucketName: "bucket", BucketName: "bucket",
Prefix: metabase.ObjectKey(prefixes[i]), Prefix: metabase.ObjectKey(prefixes[i]),
Status: metabase.CommittedUnversioned, Pending: false,
}, func(ctx context.Context, it metabase.ObjectsIterator) error { }, func(ctx context.Context, it metabase.ObjectsIterator) error {
var entry metabase.ObjectEntry var entry metabase.ObjectEntry
for it.Next(ctx, &entry) { for it.Next(ctx, &entry) {
@ -333,7 +333,7 @@ func (s *scenario) run(ctx *testcontext.Context, b *testing.B, db *metabase.DB)
Cursor: metabase.IterateCursor{ Cursor: metabase.IterateCursor{
Key: object.ObjectKey, Key: object.ObjectKey,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, func(ctx context.Context, it metabase.ObjectsIterator) error { }, func(ctx context.Context, it metabase.ObjectsIterator) error {

6
satellite/metabase/get.go
View File

@ -336,10 +336,10 @@ func (db *DB) BucketEmpty(ctx context.Context, opts BucketEmpty) (empty bool, er
func (db *DB) TestingAllCommittedObjects(ctx context.Context, projectID uuid.UUID, bucketName string) (objects []ObjectEntry, err error) { func (db *DB) TestingAllCommittedObjects(ctx context.Context, projectID uuid.UUID, bucketName string) (objects []ObjectEntry, err error) {
defer mon.Task()(&ctx)(&err) defer mon.Task()(&ctx)(&err)
return db.testingAllObjectsByStatus(ctx, projectID, bucketName, CommittedUnversioned) return db.testingAllObjectsByStatus(ctx, projectID, bucketName, false)
} }
func (db *DB) testingAllObjectsByStatus(ctx context.Context, projectID uuid.UUID, bucketName string, status ObjectStatus) (objects []ObjectEntry, err error) { func (db *DB) testingAllObjectsByStatus(ctx context.Context, projectID uuid.UUID, bucketName string, pending bool) (objects []ObjectEntry, err error) {
defer mon.Task()(&ctx)(&err) defer mon.Task()(&ctx)(&err)
err = db.IterateObjectsAllVersionsWithStatus(ctx, err = db.IterateObjectsAllVersionsWithStatus(ctx,
@ -347,7 +347,7 @@ func (db *DB) testingAllObjectsByStatus(ctx context.Context, projectID uuid.UUID
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: status, Pending: pending,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, func(ctx context.Context, it ObjectsIterator) error { }, func(ctx context.Context, it ObjectsIterator) error {

43
satellite/metabase/iterator.go
View File

@ -20,7 +20,7 @@ type objectsIterator struct {
projectID uuid.UUID projectID uuid.UUID
bucketName []byte bucketName []byte
status ObjectStatus pending bool
prefix ObjectKey prefix ObjectKey
prefixLimit ObjectKey prefixLimit ObjectKey
batchSize int batchSize int
@ -54,7 +54,7 @@ func iterateAllVersionsWithStatus(ctx context.Context, db *DB, opts IterateObjec
projectID: opts.ProjectID, projectID: opts.ProjectID,
bucketName: []byte(opts.BucketName), bucketName: []byte(opts.BucketName),
status: opts.Status, pending: opts.Pending,
prefix: opts.Prefix, prefix: opts.Prefix,
prefixLimit: prefixLimit(opts.Prefix), prefixLimit: prefixLimit(opts.Prefix),
batchSize: opts.BatchSize, batchSize: opts.BatchSize,
@ -98,7 +98,7 @@ func iteratePendingObjectsByKey(ctx context.Context, db *DB, opts IteratePending
recursive: true, recursive: true,
includeCustomMetadata: true, includeCustomMetadata: true,
includeSystemMetadata: true, includeSystemMetadata: true,
status: Pending, pending: true,
curIndex: 0, curIndex: 0,
cursor: iterateCursor{ cursor: iterateCursor{
@ -162,7 +162,12 @@ func (it *objectsIterator) Next(ctx context.Context, item *ObjectEntry) bool {
*item = ObjectEntry{ *item = ObjectEntry{
IsPrefix: true, IsPrefix: true,
ObjectKey: item.ObjectKey[:p+1], ObjectKey: item.ObjectKey[:p+1],
Status: it.status, }
// TODO(ver): should we return something else here?
if it.pending {
item.Status = Pending
} else {
item.Status = CommittedUnversioned
} }
} }
@ -231,6 +236,11 @@ func doNextQueryAllVersionsWithStatus(ctx context.Context, it *objectsIterator)
cursorCompare = ">=" cursorCompare = ">="
} }
statusFilter := `AND status <> ` + statusPending
if it.pending {
statusFilter = `AND status = ` + statusPending
}
if it.prefixLimit == "" { if it.prefixLimit == "" {
querySelectFields := querySelectorFields("object_key", it) querySelectFields := querySelectorFields("object_key", it)
return it.db.db.QueryContext(ctx, ` return it.db.db.QueryContext(ctx, `
@ -238,14 +248,13 @@ func doNextQueryAllVersionsWithStatus(ctx context.Context, it *objectsIterator)
`+querySelectFields+` `+querySelectFields+`
FROM objects FROM objects
WHERE WHERE
(project_id, bucket_name, object_key, version) `+cursorCompare+` ($1, $2, $4, $5) (project_id, bucket_name, object_key, version) `+cursorCompare+` ($1, $2, $3, $4)
AND (project_id, bucket_name) < ($1, $7) AND (project_id, bucket_name) < ($1, $6)
AND status = $3 `+statusFilter+`
AND (expires_at IS NULL OR expires_at > now()) AND (expires_at IS NULL OR expires_at > now())
ORDER BY (project_id, bucket_name, object_key, version) ASC ORDER BY (project_id, bucket_name, object_key, version) ASC
LIMIT $6 LIMIT $5
`, it.projectID, it.bucketName, `, it.projectID, it.bucketName,
it.status,
[]byte(it.cursor.Key), int(it.cursor.Version), []byte(it.cursor.Key), int(it.cursor.Version),
it.batchSize, it.batchSize,
nextBucket(it.bucketName), nextBucket(it.bucketName),
@ -257,20 +266,19 @@ func doNextQueryAllVersionsWithStatus(ctx context.Context, it *objectsIterator)
fromSubstring = len(it.prefix) + 1 fromSubstring = len(it.prefix) + 1
} }
querySelectFields := querySelectorFields("SUBSTRING(object_key FROM $8)", it) querySelectFields := querySelectorFields("SUBSTRING(object_key FROM $7)", it)
return it.db.db.QueryContext(ctx, ` return it.db.db.QueryContext(ctx, `
SELECT SELECT
`+querySelectFields+` `+querySelectFields+`
FROM objects FROM objects
WHERE WHERE
(project_id, bucket_name, object_key, version) `+cursorCompare+` ($1, $2, $4, $5) (project_id, bucket_name, object_key, version) `+cursorCompare+` ($1, $2, $3, $4)
AND (project_id, bucket_name, object_key) < ($1, $2, $6) AND (project_id, bucket_name, object_key) < ($1, $2, $5)
AND status = $3 `+statusFilter+`
AND (expires_at IS NULL OR expires_at > now()) AND (expires_at IS NULL OR expires_at > now())
ORDER BY (project_id, bucket_name, object_key, version) ASC ORDER BY (project_id, bucket_name, object_key, version) ASC
LIMIT $7 LIMIT $6
`, it.projectID, it.bucketName, `, it.projectID, it.bucketName,
it.status,
[]byte(it.cursor.Key), int(it.cursor.Version), []byte(it.cursor.Key), int(it.cursor.Version),
[]byte(it.prefixLimit), []byte(it.prefixLimit),
it.batchSize, it.batchSize,
@ -282,6 +290,7 @@ func querySelectorFields(objectKeyColumn string, it *objectsIterator) string {
querySelectFields := objectKeyColumn + ` querySelectFields := objectKeyColumn + `
,stream_id ,stream_id
,version ,version
,status
,encryption` ,encryption`
if it.includeSystemMetadata { if it.includeSystemMetadata {
@ -317,7 +326,7 @@ func doNextQueryPendingObjectsByKey(ctx context.Context, it *objectsIterator) (_
return it.db.db.QueryContext(ctx, ` return it.db.db.QueryContext(ctx, `
SELECT SELECT
object_key, stream_id, version, encryption, object_key, stream_id, version, status, encryption,
created_at, expires_at, created_at, expires_at,
segment_count, segment_count,
total_plain_size, total_encrypted_size, fixed_segment_size, total_plain_size, total_encrypted_size, fixed_segment_size,
@ -339,12 +348,12 @@ func doNextQueryPendingObjectsByKey(ctx context.Context, it *objectsIterator) (_
// scanItem scans doNextQuery results into ObjectEntry. // scanItem scans doNextQuery results into ObjectEntry.
func (it *objectsIterator) scanItem(item *ObjectEntry) (err error) { func (it *objectsIterator) scanItem(item *ObjectEntry) (err error) {
item.IsPrefix = false item.IsPrefix = false
item.Status = it.status
fields := []interface{}{ fields := []interface{}{
&item.ObjectKey, &item.ObjectKey,
&item.StreamID, &item.StreamID,
&item.Version, &item.Version,
&item.Status,
encryptionParameters{&item.Encryption}, encryptionParameters{&item.Encryption},
} }

112
satellite/metabase/iterator_test.go
View File

@ -29,7 +29,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: uuid.UUID{}, ProjectID: uuid.UUID{},
BucketName: "sj://mybucket", BucketName: "sj://mybucket",
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
}, },
ErrClass: &metabase.ErrInvalidRequest, ErrClass: &metabase.ErrInvalidRequest,
ErrText: "ProjectID missing", ErrText: "ProjectID missing",
@ -41,7 +41,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: uuid.UUID{1}, ProjectID: uuid.UUID{1},
BucketName: "", BucketName: "",
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
}, },
ErrClass: &metabase.ErrInvalidRequest, ErrClass: &metabase.ErrInvalidRequest,
ErrText: "BucketName missing", ErrText: "BucketName missing",
@ -54,24 +54,12 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: "mybucket", BucketName: "mybucket",
BatchSize: -1, BatchSize: -1,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
}, },
ErrClass: &metabase.ErrInvalidRequest, ErrClass: &metabase.ErrInvalidRequest,
ErrText: "BatchSize is negative", ErrText: "BatchSize is negative",
}.Check(ctx, t, db) }.Check(ctx, t, db)
}) })
t.Run("Status is invalid", func(t *testing.T) {
metabasetest.IterateObjectsWithStatus{
Opts: metabase.IterateObjectsWithStatus{
ProjectID: uuid.UUID{1},
BucketName: "test",
Recursive: true,
Status: 255,
},
ErrClass: &metabase.ErrInvalidRequest,
ErrText: "Status 255 is not supported",
}.Check(ctx, t, db)
})
}) })
t.Run("empty bucket", func(t *testing.T) { t.Run("empty bucket", func(t *testing.T) {
@ -83,7 +71,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: "myemptybucket", BucketName: "myemptybucket",
BatchSize: 10, BatchSize: 10,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
}, },
Result: nil, Result: nil,
}.Check(ctx, t, db) }.Check(ctx, t, db)
@ -135,7 +123,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -157,7 +145,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.Pending, Pending: true,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -187,7 +175,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: "mybucket", BucketName: "mybucket",
Recursive: true, Recursive: true,
BatchSize: limit, BatchSize: limit,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -211,7 +199,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: "mybucket", BucketName: "mybucket",
Recursive: true, Recursive: true,
BatchSize: limit, BatchSize: limit,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -238,7 +226,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: uuid.UUID{1}, ProjectID: uuid.UUID{1},
BucketName: "bucket-a", BucketName: "bucket-a",
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -267,7 +255,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: uuid.UUID{1}, ProjectID: uuid.UUID{1},
BucketName: "mybucket", BucketName: "mybucket",
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -300,7 +288,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -322,7 +310,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -345,7 +333,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -368,7 +356,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -386,7 +374,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -405,7 +393,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -423,7 +411,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -454,7 +442,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -471,7 +459,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -489,7 +477,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -507,7 +495,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -524,7 +512,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -542,7 +530,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -559,7 +547,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -573,7 +561,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -591,7 +579,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Opts: metabase.IterateObjectsWithStatus{ Opts: metabase.IterateObjectsWithStatus{
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
@ -635,7 +623,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Version: -1, Version: -1,
}, },
Prefix: prefix, Prefix: prefix,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
}, collector.Add) }, collector.Add)
require.NoError(t, err) require.NoError(t, err)
@ -650,7 +638,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
}, },
Prefix: prefix, Prefix: prefix,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
}, collector.Add) }, collector.Add)
require.NoError(t, err) require.NoError(t, err)
@ -672,7 +660,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
Version: -1, Version: -1,
}, },
Prefix: metabase.ObjectKey([]byte{1}), Prefix: metabase.ObjectKey([]byte{1}),
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, collector.Add) }, collector.Add)
@ -695,7 +683,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: bucketName, BucketName: bucketName,
Prefix: metabase.ObjectKey("a/"), Prefix: metabase.ObjectKey("a/"),
BatchSize: 1, BatchSize: 1,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
}, collector.Add) }, collector.Add)
require.NoError(t, err) require.NoError(t, err)
@ -721,7 +709,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: obj1.ProjectID, ProjectID: obj1.ProjectID,
BucketName: obj1.BucketName, BucketName: obj1.BucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, collector.Add) }, collector.Add)
@ -754,7 +742,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: obj1.ProjectID, ProjectID: obj1.ProjectID,
BucketName: obj1.BucketName, BucketName: obj1.BucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: false, IncludeCustomMetadata: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, collector.Add) }, collector.Add)
@ -788,7 +776,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: obj1.ProjectID, ProjectID: obj1.ProjectID,
BucketName: obj1.BucketName, BucketName: obj1.BucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: false, IncludeSystemMetadata: false,
}, collector.Add) }, collector.Add)
@ -830,7 +818,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Prefix: metabase.ObjectKey("a/"), Prefix: metabase.ObjectKey("a/"),
Status: metabase.CommittedUnversioned, Pending: false,
BatchSize: 1, BatchSize: 1,
}, collector.Add) }, collector.Add)
require.NoError(t, err) require.NoError(t, err)
@ -891,7 +879,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: stream.ProjectID, ProjectID: stream.ProjectID,
BucketName: stream.BucketName, BucketName: stream.BucketName,
BatchSize: batchSize, BatchSize: batchSize,
Status: 3, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
} }
metabasetest.IterateObjectsWithStatus{ metabasetest.IterateObjectsWithStatus{
@ -927,7 +915,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
BucketName: bucketName, BucketName: bucketName,
Recursive: false, Recursive: false,
Prefix: "aaaa/", Prefix: "aaaa/",
Status: metabase.CommittedUnversioned, Pending: false,
BatchSize: 2, BatchSize: 2,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -974,7 +962,7 @@ func TestIterateObjectsWithStatus(t *testing.T) {
ProjectID: projectID, ProjectID: projectID,
BucketName: bucketName, BucketName: bucketName,
Recursive: true, Recursive: true,
Status: metabase.CommittedUnversioned, Pending: false,
BatchSize: 3, BatchSize: 3,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -1010,7 +998,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("08/"), Key: metabase.ObjectKey("08/"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -1030,7 +1018,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("08"), Key: metabase.ObjectKey("08"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
Result: []metabase.ObjectEntry{ Result: []metabase.ObjectEntry{
@ -1050,7 +1038,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("08/a/x"), Key: metabase.ObjectKey("08/a/x"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
Result: []metabase.ObjectEntry{ Result: []metabase.ObjectEntry{
@ -1079,7 +1067,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08"), Key: metabase.ObjectKey("2017/05/08"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
Result: []metabase.ObjectEntry{ Result: []metabase.ObjectEntry{
@ -1099,7 +1087,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08/"), Key: metabase.ObjectKey("2017/05/08/"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
Result: []metabase.ObjectEntry{ Result: []metabase.ObjectEntry{
@ -1118,7 +1106,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08/a/x"), Key: metabase.ObjectKey("2017/05/08/a/x"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
Result: []metabase.ObjectEntry{ Result: []metabase.ObjectEntry{
@ -1164,7 +1152,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08"), Key: metabase.ObjectKey("2017/05/08"),
Version: objects["2017/05/08"].Version, Version: objects["2017/05/08"].Version,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -1187,7 +1175,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08/"), Key: metabase.ObjectKey("2017/05/08/"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -1208,7 +1196,7 @@ func TestIterateObjectsSkipCursor(t *testing.T) {
Key: metabase.ObjectKey("2017/05/08/a/x"), Key: metabase.ObjectKey("2017/05/08/a/x"),
Version: 1, Version: 1,
}, },
Status: metabase.CommittedUnversioned, Pending: false,
IncludeCustomMetadata: true, IncludeCustomMetadata: true,
IncludeSystemMetadata: true, IncludeSystemMetadata: true,
}, },
@ -1466,6 +1454,8 @@ func TestIteratePendingObjectsWithObjectKey(t *testing.T) {
metabasetest.Verify{Objects: objects}.Check(ctx, t, db) metabasetest.Verify{Objects: objects}.Check(ctx, t, db)
}) })
// TODO(ver): add tests for delete markers and versioned/unversioned
}) })
} }
@ -1584,7 +1574,7 @@ func BenchmarkNonRecursiveListing(b *testing.B) {
ProjectID: baseObj.ProjectID, ProjectID: baseObj.ProjectID,
BucketName: baseObj.BucketName, BucketName: baseObj.BucketName,
BatchSize: 5, BatchSize: 5,
Status: metabase.CommittedUnversioned, Pending: false,
}, func(ctx context.Context, oi metabase.ObjectsIterator) error { }, func(ctx context.Context, oi metabase.ObjectsIterator) error {
entry := metabase.ObjectEntry{} entry := metabase.ObjectEntry{}
for oi.Next(ctx, &entry) { for oi.Next(ctx, &entry) {
@ -1602,7 +1592,7 @@ func BenchmarkNonRecursiveListing(b *testing.B) {
BucketName: baseObj.BucketName, BucketName: baseObj.BucketName,
Prefix: "foo/", Prefix: "foo/",
BatchSize: 5, BatchSize: 5,
Status: metabase.CommittedUnversioned, Pending: false,
}, func(ctx context.Context, oi metabase.ObjectsIterator) error { }, func(ctx context.Context, oi metabase.ObjectsIterator) error {
entry := metabase.ObjectEntry{} entry := metabase.ObjectEntry{}
for oi.Next(ctx, &entry) { for oi.Next(ctx, &entry) {
@ -1620,7 +1610,7 @@ func BenchmarkNonRecursiveListing(b *testing.B) {
BucketName: baseObj.BucketName, BucketName: baseObj.BucketName,
Prefix: "boo/", Prefix: "boo/",
BatchSize: 5, BatchSize: 5,
Status: metabase.CommittedUnversioned, Pending: false,
}, func(ctx context.Context, oi metabase.ObjectsIterator) error { }, func(ctx context.Context, oi metabase.ObjectsIterator) error {
entry := metabase.ObjectEntry{} entry := metabase.ObjectEntry{}
for oi.Next(ctx, &entry) { for oi.Next(ctx, &entry) {

4
satellite/metabase/list.go
View File

@ -69,7 +69,7 @@ type IterateObjectsWithStatus struct {
BatchSize int BatchSize int
Prefix ObjectKey Prefix ObjectKey
Cursor IterateCursor Cursor IterateCursor
Status ObjectStatus Pending bool
IncludeCustomMetadata bool IncludeCustomMetadata bool
IncludeSystemMetadata bool IncludeSystemMetadata bool
} }
@ -92,8 +92,6 @@ func (opts *IterateObjectsWithStatus) Verify() error {
return ErrInvalidRequest.New("BucketName missing") return ErrInvalidRequest.New("BucketName missing")
case opts.BatchSize < 0: case opts.BatchSize < 0:
return ErrInvalidRequest.New("BatchSize is negative") return ErrInvalidRequest.New("BatchSize is negative")
case !(opts.Status == Pending || opts.Status == CommittedUnversioned):
return ErrInvalidRequest.New("Status %v is not supported", opts.Status)
} }
return nil return nil
} }

4
satellite/metainfo/endpoint_object.go
View File

@ -1001,7 +1001,7 @@ func (endpoint *Endpoint) ListObjects(ctx context.Context, req *pb.ObjectListReq
}, },
Recursive: req.Recursive, Recursive: req.Recursive,
BatchSize: limit + 1, BatchSize: limit + 1,
Status: metabase.Pending, Pending: true,
IncludeCustomMetadata: includeCustomMetadata, IncludeCustomMetadata: includeCustomMetadata,
IncludeSystemMetadata: includeSystemMetadata, IncludeSystemMetadata: includeSystemMetadata,
}, func(ctx context.Context, it metabase.ObjectsIterator) error { }, func(ctx context.Context, it metabase.ObjectsIterator) error {
@ -1063,7 +1063,7 @@ func (endpoint *Endpoint) ListObjects(ctx context.Context, req *pb.ObjectListReq
}, },
Recursive: req.Recursive, Recursive: req.Recursive,
BatchSize: limit + 1, BatchSize: limit + 1,
Status: status, Pending: status == metabase.Pending,
IncludeCustomMetadata: includeCustomMetadata, IncludeCustomMetadata: includeCustomMetadata,
IncludeSystemMetadata: includeSystemMetadata, IncludeSystemMetadata: includeSystemMetadata,
}, func(ctx context.Context, it metabase.ObjectsIterator) error { }, func(ctx context.Context, it metabase.ObjectsIterator) error {