S3: deprecate PrefixFolders and introduce HideFolders

backends/s3/s3.go

@@ -78,8 +78,22 @@ type Options struct {
 
 	// PrefixFolders can be enabled to make List operations show nested prefixes as folders
 	// instead of recursively listing all contents of nested prefixes
+	//
+	// Deprecated: This option does not reflect our desire to treat blob names as keys.
+	// Please do not use it.
 	PrefixFolders bool `yaml:"prefix_folders"`
 
+	// HideFolders is an S3-specific optimization, allowing to hide all keys that
+	// have a separator '/' in their names.
+	// In case a prefix representing a folder is provided to List,
+	// that folder will be explored, and its subfolders hidden.
+	//
+	// Moreover, please note that regardless of this option,
+	// working with folders with S3 is flaky,
+	// because a `foo` key will shadow all `foo/*` keys while listing,
+	// even though those `foo/*` keys exist and they hold the values they're expected to.
+	HideFolders bool `yaml:"hide_folders"`
+
 	// EndpointURL can be set to something like "http://localhost:9000" when using Minio
 	// or "https://s3.amazonaws.com" for AWS S3.
 	EndpointURL string `yaml:"endpoint_url"`
@@ -195,7 +209,7 @@ func (b *Backend) doList(ctx context.Context, prefix string) (simpleblob.BlobLis
 
 	objCh := b.client.ListObjects(ctx, b.opt.Bucket, minio.ListObjectsOptions{
 		Prefix:    prefix,
-		Recursive: !b.opt.PrefixFolders,
+		Recursive: !b.opt.PrefixFolders && !b.opt.HideFolders,
 	})
 	for obj := range objCh {
 		// Handle error returned by MinIO client
@@ -210,6 +224,10 @@ func (b *Backend) doList(ctx context.Context, prefix string) (simpleblob.BlobLis
 			continue
 		}
 
+		if b.opt.HideFolders && strings.HasSuffix(obj.Key, "/") {
+			continue
+		}
+
 		// Strip global prefix from blob
 		blobName := obj.Key
 		if gpEndIndex > 0 {

backends/s3/s3_test.go

@@ -127,6 +127,8 @@ func TestBackend_globalPrefixAndMarker(t *testing.T) {
 }
 
 func TestBackend_recursive(t *testing.T) {
+	// NB: Those tests are for PrefixFolders, a deprecated option.
+
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
@@ -166,3 +168,40 @@ func TestBackend_recursive(t *testing.T) {
 
 	assert.Len(t, b.lastMarker, 0)
 }
+
+func TestHideFolders(t *testing.T) {
+	// NB: working with folders with S3 is flaky, because a `foo` key
+	// will shadow all `foo/*` keys while listing,
+	// even though those `foo/*` keys exist and they hold the values they're expected to.
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	b := getBackend(ctx, t)
+	b.opt.HideFolders = true
+
+	p := []byte("123")
+	err := b.Store(ctx, "foo", p)
+	assert.NoError(t, err)
+	err = b.Store(ctx, "bar/baz", p)
+	assert.NoError(t, err)
+
+	t.Run("at root", func(t *testing.T) {
+		ls, err := b.List(ctx, "")
+		assert.NoError(t, err)
+		assert.Equal(t, []string{"foo"}, ls.Names())
+	})
+
+	t.Run("with List prefix", func(t *testing.T) {
+		ls, err := b.List(ctx, "bar/")
+		assert.NoError(t, err)
+		assert.Equal(t, []string{"bar/baz"}, ls.Names())
+	})
+
+	t.Run("with global prefix", func(t *testing.T) {
+		b.setGlobalPrefix("bar/")
+		ls, err := b.List(ctx, "")
+		assert.NoError(t, err)
+		assert.Equal(t, []string{"baz"}, ls.Names())
+	})
+}