From 9c65168312da64cf9e3265bf331e39afd9ed52a9 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Wed, 18 Aug 2021 07:40:53 -0700
Subject: [PATCH] fix: all levels deep flat key match (#12996)

this addresses a regression from #12984
which only addresses flat key from single
level deep at bucket level.

added extra tests as well to cover all
these scenarios.
---
 cmd/metacache-walk.go              | 15 ++++++++++-----
 cmd/object-api-listobjects_test.go | 25 +++++++++++++++++++++++++
 2 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/cmd/metacache-walk.go b/cmd/metacache-walk.go
index 622f7a3e7..2a0288a1e 100644
--- a/cmd/metacache-walk.go
+++ b/cmd/metacache-walk.go
@@ -110,11 +110,6 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 	var scanDir func(path string) error
 
 	scanDir = func(current string) error {
-		// always skip the directory that doesn't match the prefix
-		if len(current) > 0 && !strings.HasPrefix(current, prefix) {
-			return nil
-		}
-
 		// Skip forward, if requested...
 		forward := ""
 		if len(opts.ForwardTo) > 0 && strings.HasPrefix(opts.ForwardTo, current) {
@@ -143,7 +138,16 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 		}
 		dirObjects := make(map[string]struct{})
 		for i, entry := range entries {
+			if len(prefix) > 0 && !strings.HasPrefix(entry, prefix) {
+				// Do do not retain the file, since it doesn't
+				// match the prefix.
+				entries[i] = ""
+				continue
+			}
 			if len(forward) > 0 && entry < forward {
+				// Do do not retain the file, since its
+				// lexially smaller than 'forward'
+				entries[i] = ""
 				continue
 			}
 			if strings.HasSuffix(entry, slashSeparator) {
@@ -199,6 +203,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 		// Process in sort order.
 		sort.Strings(entries)
 		dirStack := make([]string, 0, 5)
+		prefix = "" // Remove prefix after first level as we have already filtered the list.
 		if len(forward) > 0 {
 			idx := sort.SearchStrings(entries, forward)
 			if idx > 0 {
diff --git a/cmd/object-api-listobjects_test.go b/cmd/object-api-listobjects_test.go
index ec892b7ea..b5d1f2a3e 100644
--- a/cmd/object-api-listobjects_test.go
+++ b/cmd/object-api-listobjects_test.go
@@ -84,6 +84,10 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{testBuckets[4], "file1/guidSplunk-aaaa/file", "content", nil},
 		{testBuckets[5], "dir/day_id=2017-10-10/issue", "content", nil},
 		{testBuckets[5], "dir/day_id=2017-10-11/issue", "content", nil},
+		{testBuckets[5], "foo/201910/1122", "content", nil},
+		{testBuckets[5], "foo/201910/1112", "content", nil},
+		{testBuckets[5], "foo/201910/2112", "content", nil},
+		{testBuckets[5], "foo/201910_txt", "content", nil},
 	}
 	for _, object := range testObjects {
 		md5Bytes := md5.Sum([]byte(object.content))
@@ -477,6 +481,24 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 			IsTruncated: true,
 			Prefixes:    []string{"dir/day_id=2017-10-10/"},
 		},
+		// ListObjectsResult-37 list with prefix match 2 levels deep
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "foo/201910/1112"},
+				{Name: "foo/201910/1122"},
+			},
+		},
+		// ListObjectsResult-38 list with prefix match 1 level deep
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "foo/201910/1112"},
+				{Name: "foo/201910/1122"},
+				{Name: "foo/201910/2112"},
+				{Name: "foo/201910_txt"},
+			},
+		},
 	}
 
 	testCases := []struct {
@@ -602,6 +624,9 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{testBuckets[4], "file1/", "", "guidSplunk", 1000, resultCases[35], nil, true},
 		// Test listing at prefix with expected prefix markers
 		{testBuckets[5], "dir/", "", SlashSeparator, 1, resultCases[36], nil, true},
+		// Test listing with prefix match
+		{testBuckets[5], "foo/201910/11", "", "", 1000, resultCases[37], nil, true},
+		{testBuckets[5], "foo/201910", "", "", 1000, resultCases[38], nil, true},
 	}
 
 	for i, testCase := range testCases {