satellite/metabase: jump after prefix at the end of iteration page

For non-recursive listing if cursor to the next iteration page
is a prefix then jump after this prefix to avoid listing objects from
this prefix.

Benchmark against 'main' - 7e5025cac0
name                                                 old time/op    new time/op    delta
NonRecursiveListing/Cockroach/listing_no_prefix-8      11.7ms ±29%     1.7ms ± 7%  -85.52%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_with_prefix-8    4.60ms ± 6%    3.23ms ± 3%  -29.72%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_only_prefix-8    7.38ms ± 7%    8.47ms ±34%     ~     (p=0.421 n=5+5)

name                                                 old alloc/op   new alloc/op   delta
NonRecursiveListing/Cockroach/listing_no_prefix-8      92.3kB ± 0%    16.9kB ± 0%  -81.72%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_with_prefix-8    38.0kB ± 0%    27.4kB ± 0%  -28.02%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_only_prefix-8    59.8kB ± 0%    60.1kB ± 0%   +0.38%  (p=0.008 n=5+5)

name                                                 old allocs/op  new allocs/op  delta
NonRecursiveListing/Cockroach/listing_no_prefix-8       1.78k ± 0%     0.31k ± 0%  -82.48%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_with_prefix-8       734 ± 0%       526 ± 0%  -28.34%  (p=0.008 n=5+5)
NonRecursiveListing/Cockroach/listing_only_prefix-8     1.15k ± 0%     1.16k ± 0%   +0.87%  (p=0.008 n=5+5)

Discussed here https://github.com/storj/team-metainfo/issues/116

Change-Id: Iff671f062d9af83ec419334089163dd204d0cac0

satellite/metabase/iterator.go

@@ -138,10 +138,6 @@ func (it *objectsIterator) Next(ctx context.Context, item *ObjectEntry) bool {
 	}
 
 	// TODO: implement this on the database side
-	ok := it.next(ctx, item)
-	if !ok {
-		return false
-	}
 
 	// skip until we are past the prefix we returned before.
 	if it.skipPrefix != "" {
@@ -151,6 +147,11 @@ func (it *objectsIterator) Next(ctx context.Context, item *ObjectEntry) bool {
 			}
 		}
 		it.skipPrefix = ""
+	} else {
+		ok := it.next(ctx, item)
+		if !ok {
+			return false
+		}
 	}
 
 	// should this be treated as a prefix?
@@ -179,6 +180,16 @@ func (it *objectsIterator) next(ctx context.Context, item *ObjectEntry) bool {
 			return false
 		}
 
+		if !it.recursive {
+			afterPrefix := it.cursor.Key[len(it.prefix):]
+			p := bytes.IndexByte([]byte(afterPrefix), Delimiter)
+			if p >= 0 {
+				it.cursor.Key = it.prefix + prefixLimit(afterPrefix[:p+1])
+				it.cursor.StreamID = uuid.UUID{}
+				it.cursor.Version = 0
+			}
+		}
+
 		rows, err := it.doNextQuery(ctx, it)
 		if err != nil {
 			it.failErr = errs.Combine(it.failErr, err)

satellite/metabase/iterator_test.go

@@ -910,6 +910,34 @@ func TestIterateObjectsWithStatus(t *testing.T) {
 				})
 			}
 		})
+
+		t.Run("prefix longer than key", func(t *testing.T) {
+			defer metabasetest.DeleteAll{}.Check(ctx, t, db)
+
+			projectID, bucketName := uuid.UUID{1}, "bucky"
+			objects := createObjectsWithKeys(ctx, t, db, projectID, bucketName, []metabase.ObjectKey{
+				"aaaa/a",
+				"aaaa/b",
+				"aaaa/c",
+			})
+
+			metabasetest.IterateObjectsWithStatus{
+				Opts: metabase.IterateObjectsWithStatus{
+					ProjectID:             projectID,
+					BucketName:            bucketName,
+					Recursive:             false,
+					Prefix:                "aaaa/",
+					Status:                metabase.Committed,
+					BatchSize:             2,
+					IncludeSystemMetadata: true,
+				},
+				Result: withoutPrefix("aaaa/",
+					objects["aaaa/a"],
+					objects["aaaa/b"],
+					objects["aaaa/c"],
+				),
+			}.Check(ctx, t, db)
+		})
 	})
 }
 
@@ -1442,6 +1470,11 @@ func BenchmarkNonRecursiveListing(b *testing.B) {
 			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
 		}
 
+		for i := 0; i < 50; i++ {
+			baseObj.ObjectKey = metabase.ObjectKey("boo/foo" + strconv.Itoa(i) + "/object")
+			metabasetest.CreateObject(ctx, b, db, baseObj, 0)
+		}
+
 		b.Run("listing no prefix", func(b *testing.B) {
 			for i := 0; i < b.N; i++ {
 				err := db.IterateObjectsAllVersionsWithStatus(ctx, metabase.IterateObjectsWithStatus{
@@ -1476,5 +1509,23 @@ func BenchmarkNonRecursiveListing(b *testing.B) {
 				require.NoError(b, err)
 			}
 		})
+
+		b.Run("listing only prefix", func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				err := db.IterateObjectsAllVersionsWithStatus(ctx, metabase.IterateObjectsWithStatus{
+					ProjectID:  baseObj.ProjectID,
+					BucketName: baseObj.BucketName,
+					Prefix:     "boo/",
+					BatchSize:  5,
+					Status:     metabase.Committed,
+				}, func(ctx context.Context, oi metabase.ObjectsIterator) error {
+					entry := metabase.ObjectEntry{}
+					for oi.Next(ctx, &entry) {
+					}
+					return nil
+				})
+				require.NoError(b, err)
+			}
+		})
 	})
 }