Repository: minio/minio (language: go)

Issue:
optimize max-keys=2 listing for spark workloads
to return results appropriately for versioned buckets, especially when underlying prefixes have been deleted

## Community Contribution License
All community contributions in this pull request are licensed to the project maintainers
under the terms of the [Apache 2 license](https://www.apache.org/licenses/LICENSE-2.0). 
By creating this pull request I represent that I have the right to license the 
contributions to the project maintainers under the Apache 2 license.

## Description


## Motivation and Context
solve partitioned overwrite issues in versioned buckets for spark/hadoop workloads using minio as object store  - prefixes in deleted state should not be returned in the result for calls similar to this:
```
 GET /warehouse/?list-type=2&delimiter=%2F&max-keys=2&prefix=partition%2Fnyc_taxis%2Fdata.parquet%2F_temporary%2F0%2Ftask_202405120118115803861317163525099_0000_m_000000%2F&fetch-owner=false
```

## How to test this PR?
run a spark+hive+minio job on a versioned bucket
https://github.com/minio/openlake/tree/main/spark/partition-test

## Types of changes
- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [x] Optimization (provides speedup with no functional changes)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Checklist:
- [ ] Fixes a regression (If yes, please add `commit-id` or `PR #` here)
- [ ] Unit tests added/updated
- [ ] Internal documentation updated
- [ ] Create a documentation update request [here](https://github.com/minio/docs/issues/new?label=doc-change,title=Doc+Updated+Needed+For+PR+github.com%2fminio%2fminio%2fpull%2fNNNNN)



Discussion / hints:
Can we run the code when we are running mint tests? 
> Can we run the code when we are running mint tests?

we may be able to do this in minio-go by setting a custom user agent
> we may be able to do this in minio-go by setting a custom user agent

okay cool, we should do that @poornas IMO. 

Fix patch:
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 91384f51e1eeb..b8ecad0de7096 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1647,12 +1647,17 @@ func (z *erasureServerPools) listObjectsGeneric(ctx context.Context, bucket, pre
 						}
 					}
 				}
-				if o.DeleteMarker {
+				switch {
+				case o.DeleteMarker:
 					loi.Objects = append(loi.Objects, ObjectInfo{Bucket: bucket, IsDir: true, Name: prefix})
 					return loi, nil
-				} else if len(li.Objects) == 1 {
+				case len(li.Objects) >= 1:
 					loi.Objects = append(loi.Objects, o)
-					loi.Prefixes = append(loi.Prefixes, path.Join(opts.Prefix, pfx))
+					if pfx != "" {
+						loi.Prefixes = append(loi.Prefixes, path.Join(opts.Prefix, pfx))
+					}
+				case len(li.Prefixes) > 0:
+					loi.Prefixes = append(loi.Prefixes, li.Prefixes...)
 				}
 			}
 			return loi, nil


Test patch:
diff --git a/cmd/server_test.go b/cmd/server_test.go
index 1006637022c55..037a3ea951515 100644
--- a/cmd/server_test.go
+++ b/cmd/server_test.go
@@ -1671,7 +1671,7 @@ func (s *TestSuiteCommon) TestListObjectsV2HadoopUAHandler(c *check) {
 	c.Assert(err, nil)
 	c.Assert(response.StatusCode, http.StatusOK)
 
-	for _, objectName := range []string{"pfx/a/1.txt", "pfx/b/2.txt", "pfx/", "pfx2/c/3.txt", "pfx2/d/3.txt", "pfx1/1.txt", "pfx2/"} {
+	for _, objectName := range []string{"pfx/a/1.txt", "pfx/b/2.txt", "pfx/", "pfx2/c/3.txt", "pfx2/d/3.txt", "pfx1/1.txt", "pfx2/", "pfx3/", "pfx4/"} {
 		buffer := bytes.NewReader([]byte(""))
 		request, err = newTestSignedRequest(http.MethodPut, getPutObjectURL(s.endPoint, bucketName, objectName),
 			int64(buffer.Len()), buffer, s.accessKey, s.secretKey, s.signer)
@@ -1680,7 +1680,7 @@ func (s *TestSuiteCommon) TestListObjectsV2HadoopUAHandler(c *check) {
 		c.Assert(err, nil)
 		c.Assert(response.StatusCode, http.StatusOK)
 	}
-	for _, objectName := range []string{"pfx2/c/3.txt", "pfx2/d/3.txt", "pfx2/"} {
+	for _, objectName := range []string{"pfx2/c/3.txt", "pfx2/d/3.txt", "pfx2/", "pfx3/"} {
 		delRequest, err := newTestSignedRequest(http.MethodDelete, getDeleteObjectURL(s.endPoint, bucketName, objectName),
 			0, nil, s.accessKey, s.secretKey, s.signer)
 		c.Assert(err, nil)
@@ -1697,7 +1697,6 @@ func (s *TestSuiteCommon) TestListObjectsV2HadoopUAHandler(c *check) {
 			getListObjectsV2URL(s.endPoint, bucketName, "pfx/a/", "2", "", "", "/"),
 			[]string{
 				"<Contents><Key>pfx/a/1.txt</Key>",
-				"<CommonPrefixes><Prefix>pfx/a</Prefix></CommonPrefixes>",
 			},
 			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
 		},
@@ -1720,21 +1719,24 @@ func (s *TestSuiteCommon) TestListObjectsV2HadoopUAHandler(c *check) {
 		{
 			getListObjectsV2URL(s.endPoint, bucketName, "pfx2/c", "2", "true", "", "/"),
 			[]string{
-				"<Prefix>pfx2/c</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<Prefix>pfx2/c</Prefix><KeyCount>1</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<CommonPrefixes><Prefix>pfx2/c/</Prefix>",
 			},
 			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
 		},
 		{
 			getListObjectsV2URL(s.endPoint, bucketName, "pfx2/c", "2", "true", "", "/"),
 			[]string{
-				"<Prefix>pfx2/c</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<Prefix>pfx2/c</Prefix><KeyCount>1</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<CommonPrefixes><Prefix>pfx2/c/</Prefix></CommonPrefixes>",
 			},
 			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
 		},
 		{
 			getListObjectsV2URL(s.endPoint, bucketName, "pfx2/", "2", "false", "", "/"),
 			[]string{
-				"<Prefix>pfx2/</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<Prefix>pfx2/</Prefix><KeyCount>2</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<CommonPrefixes><Prefix>pfx2/c/</Prefix></CommonPrefixes><CommonPrefixes><Prefix>pfx2/d/</Prefix></CommonPrefixes>",
 			},
 			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
 		},
@@ -1748,10 +1750,39 @@ func (s *TestSuiteCommon) TestListObjectsV2HadoopUAHandler(c *check) {
 		{
 			getListObjectsV2URL(s.endPoint, bucketName, "pfx2/", "2", "false", "", "/"),
 			[]string{
-				"<Prefix>pfx2/</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<Prefix>pfx2/</Prefix><KeyCount>2</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+				"<CommonPrefixes><Prefix>pfx2/c/</Prefix></CommonPrefixes><CommonPrefixes><Prefix>pfx2/d/</Prefix></CommonPrefixes>",
+			},
+			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
+		},
+		{
+			getListObjectsV2URL(s.endPoint, bucketName, "pfx3/", "2", "false", "", "/"),
+			[]string{
+				"<Prefix>pfx3/</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+			},
+			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
+		},
+		{
+			getListObjectsV2URL(s.endPoint, bucketName, "pfx4/", "2", "false", "", "/"),
+			[]string{
+				"<Prefix>pfx4/</Prefix><KeyCount>1</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated><Contents><Key>pfx4/</Key>",
 			},
 			"Hadoop 3.3.2, aws-sdk-java/1.12.262 Linux/5.14.0-362.24.1.el9_3.x86_64 OpenJDK_64-Bit_Server_VM/11.0.22+7 java/11.0.22 scala/2.12.15 vendor/Eclipse_Adoptium cfg/retry-mode/legacy",
 		},
+		{
+			getListObjectsV2URL(s.endPoint, bucketName, "pfx3/", "2", "false", "", "/"),
+			[]string{
+				"<Prefix>pfx3/</Prefix><KeyCount>0</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated>",
+			},
+			"",
+		},
+		{
+			getListObjectsV2URL(s.endPoint, bucketName, "pfx4/", "2", "false", "", "/"),
+			[]string{
+				"<Prefix>pfx4/</Prefix><KeyCount>1</KeyCount><MaxKeys>2</MaxKeys><Delimiter>/</Delimiter><IsTruncated>false</IsTruncated><Contents><Key>pfx4/</Key>",
+			},
+			"",
+		},
 	}
 	for _, testCase := range testCases {
 		// create listObjectsV1 request with valid parameters