List with delimiter (#2813)

juicedata · Sep 29, 2022 · 953a763 · 953a763
1 parent 7382907
commit 953a763
Show file tree

Hide file tree

Showing 35 changed files with 264 additions and 79 deletions.
diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml
@@ -30,7 +30,7 @@ jobs:
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.29
+          version: v1.49.0
   build:
     runs-on: ubuntu-latest
     steps:
@@ -79,4 +79,4 @@ jobs:
       run: |
         wget https://github.com/apple/foundationdb/releases/download/6.3.23/foundationdb-clients_6.3.23-1_amd64.deb
         sudo dpkg -i foundationdb-clients_6.3.23-1_amd64.deb
-        GOPATH=$HOME/go make juicefs.fdb
+        GOPATH=$HOME/go make juicefs.fdb
diff --git a/cmd/objbench.go b/cmd/objbench.go
@@ -562,7 +562,7 @@ func (bm *benchMarkObj) chtimes(key string, startKey int) error {
 }
 
 func listAll(s object.ObjectStorage, prefix, marker string, limit int64) ([]object.Object, error) {
-	r, err := s.List(prefix, marker, limit)
+	r, err := s.List(prefix, marker, "", limit)
 	if err == nil {
 		return r, nil
 	}

diff --git a/pkg/object/azure.go b/pkg/object/azure.go
@@ -93,7 +93,11 @@ func (b *wasb) Delete(key string) error {
 	return err
 }
 
-func (b *wasb) List(prefix, marker string, limit int64) ([]Object, error) {
+func (b *wasb) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
+	if delimiter != "" {
+		return nil, notSupportedDelimiter
+	}
+	// todo
 	if marker != "" {
 		if b.marker == "" {
 			// last page

diff --git a/pkg/object/b2.go b/pkg/object/b2.go
@@ -114,15 +114,15 @@ func (c *b2client) Delete(key string) error {
 	return err
 }
 
-func (c *b2client) List(prefix, marker string, limit int64) ([]Object, error) {
+func (c *b2client) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	if limit > 1000 {
 		limit = 1000
 	}
 	if marker == "" && c.nextMarker != "" {
 		marker = c.nextMarker
 		c.nextMarker = ""
 	}
-	resp, err := c.bucket.ListFileNamesWithPrefix(marker, int(limit), prefix, "")
+	resp, err := c.bucket.ListFileNamesWithPrefix(marker, int(limit), prefix, delimiter)
 	if err != nil {
 		return nil, err
 	}

diff --git a/pkg/object/bos.go b/pkg/object/bos.go
@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"time"
 
@@ -112,12 +113,12 @@ func (q *bosclient) Delete(key string) error {
 	return err
 }
 
-func (q *bosclient) List(prefix, marker string, limit int64) ([]Object, error) {
+func (q *bosclient) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	if limit > 1000 {
 		limit = 1000
 	}
 	limit_ := int(limit)
-	out, err := q.c.SimpleListObjects(q.bucket, prefix, limit_, marker, "")
+	out, err := q.c.SimpleListObjects(q.bucket, prefix, limit_, marker, delimiter)
 	if err != nil {
 		return nil, err
 	}
@@ -128,6 +129,12 @@ func (q *bosclient) List(prefix, marker string, limit int64) ([]Object, error) {
 		mod, _ := time.Parse("2006-01-02T15:04:05Z", k.LastModified)
 		objs[i] = &obj{k.Key, int64(k.Size), mod, strings.HasSuffix(k.Key, "/")}
 	}
+	if delimiter != "" {
+		for _, p := range out.CommonPrefixes {
+			objs = append(objs, &obj{p.Prefix, 0, time.Unix(0, 0), true})
+		}
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
+	}
 	return objs, nil
 }
 

diff --git a/pkg/object/cos.go b/pkg/object/cos.go
@@ -26,6 +26,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -119,11 +120,12 @@ func (c *COS) Delete(key string) error {
 	return err
 }
 
-func (c *COS) List(prefix, marker string, limit int64) ([]Object, error) {
+func (c *COS) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	param := cos.BucketGetOptions{
-		Prefix:  prefix,
-		Marker:  marker,
-		MaxKeys: int(limit),
+		Prefix:    prefix,
+		Marker:    marker,
+		MaxKeys:   int(limit),
+		Delimiter: delimiter,
 	}
 	resp, _, err := c.c.Bucket.Get(ctx, &param)
 	for err == nil && len(resp.Contents) == 0 && resp.IsTruncated {
@@ -140,6 +142,12 @@ func (c *COS) List(prefix, marker string, limit int64) ([]Object, error) {
 		t, _ := time.Parse(time.RFC3339, o.LastModified)
 		objs[i] = &obj{o.Key, int64(o.Size), t, strings.HasSuffix(o.Key, "/")}
 	}
+	if delimiter != "" {
+		for _, p := range resp.CommonPrefixes {
+			objs = append(objs, &obj{p, 0, time.Unix(0, 0), true})
+		}
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
+	}
 	return objs, nil
 }
 

diff --git a/pkg/object/etcd.go b/pkg/object/etcd.go
@@ -111,7 +111,10 @@ func genNextKey(key string) string {
 	return string(next)
 }
 
-func (c *etcdClient) List(prefix, marker string, limit int64) ([]Object, error) {
+func (c *etcdClient) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
+	if delimiter != "" {
+		return nil, notSupportedDelimiter
+	}
 	if marker == "" {
 		marker = prefix
 	}

diff --git a/pkg/object/file.go b/pkg/object/file.go
@@ -319,7 +319,7 @@ func readDirSorted(dirname string) ([]*mEntry, error) {
 	return mEntries, err
 }
 
-func (d *filestore) List(prefix, marker string, limit int64) ([]Object, error) {
+func (d *filestore) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	return nil, notSupported
 }
 

diff --git a/pkg/object/gs.go b/pkg/object/gs.go
@@ -25,7 +25,9 @@ import (
 	"io"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
+	"time"
 
 	"cloud.google.com/go/compute/metadata"
 	"cloud.google.com/go/storage"
@@ -48,7 +50,7 @@ func (g *gs) String() string {
 
 func (g *gs) Create() error {
 	// check if the bucket is already exists
-	if objs, err := g.List("", "", 1); err == nil && len(objs) > 0 {
+	if objs, err := g.List("", "", "", 1); err == nil && len(objs) > 0 {
 		return nil
 	}
 
@@ -135,12 +137,12 @@ func (g *gs) Delete(key string) error {
 	return nil
 }
 
-func (g *gs) List(prefix, marker string, limit int64) ([]Object, error) {
+func (g *gs) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	if marker != "" && g.pageToken == "" {
 		// last page
 		return nil, nil
 	}
-	objectIterator := g.client.Bucket(g.bucket).Objects(ctx, &storage.Query{Prefix: prefix})
+	objectIterator := g.client.Bucket(g.bucket).Objects(ctx, &storage.Query{Prefix: prefix, Delimiter: delimiter})
 	pager := iterator.NewPager(objectIterator, int(limit), g.pageToken)
 	var entries []*storage.ObjectAttrs
 	nextPageToken, err := pager.NextPage(&entries)
@@ -152,7 +154,14 @@ func (g *gs) List(prefix, marker string, limit int64) ([]Object, error) {
 	objs := make([]Object, n)
 	for i := 0; i < n; i++ {
 		item := entries[i]
-		objs[i] = &obj{item.Name, item.Size, item.Updated, strings.HasSuffix(item.Name, "/")}
+		if delimiter != "" && item.Prefix != "" {
+			objs[i] = &obj{item.Prefix, 0, time.Unix(0, 0), true}
+		} else {
+			objs[i] = &obj{item.Name, item.Size, item.Updated, strings.HasSuffix(item.Name, "/")}
+		}
+	}
+	if delimiter != "" {
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
 	}
 	return objs, nil
 }

diff --git a/pkg/object/hdfs.go b/pkg/object/hdfs.go
@@ -177,7 +177,7 @@ func (h *hdfsclient) Delete(key string) error {
 	return err
 }
 
-func (h *hdfsclient) List(prefix, marker string, limit int64) ([]Object, error) {
+func (h *hdfsclient) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	return nil, notSupported
 }
 

diff --git a/pkg/object/ibmcos.go b/pkg/object/ibmcos.go
@@ -27,7 +27,9 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
+	"time"
 
 	"github.com/IBM/ibm-cos-sdk-go/aws"
 	"github.com/IBM/ibm-cos-sdk-go/aws/awserr"
@@ -134,12 +136,13 @@ func (s *ibmcos) Delete(key string) error {
 	return err
 }
 
-func (s *ibmcos) List(prefix, marker string, limit int64) ([]Object, error) {
+func (s *ibmcos) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	param := s3.ListObjectsInput{
-		Bucket:  &s.bucket,
-		Prefix:  &prefix,
-		Marker:  &marker,
-		MaxKeys: &limit,
+		Bucket:    &s.bucket,
+		Prefix:    &prefix,
+		Marker:    &marker,
+		MaxKeys:   &limit,
+		Delimiter: &delimiter,
 	}
 	resp, err := s.s3.ListObjects(&param)
 	if err != nil {
@@ -151,6 +154,12 @@ func (s *ibmcos) List(prefix, marker string, limit int64) ([]Object, error) {
 		o := resp.Contents[i]
 		objs[i] = &obj{*o.Key, *o.Size, *o.LastModified, strings.HasSuffix(*o.Key, "/")}
 	}
+	if delimiter != "" {
+		for _, p := range resp.CommonPrefixes {
+			objs = append(objs, &obj{*p.Prefix, 0, time.Unix(0, 0), true})
+		}
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
+	}
 	return objs, nil
 }
 

diff --git a/pkg/object/interface.go b/pkg/object/interface.go
@@ -77,7 +77,7 @@ type ObjectStorage interface {
 	// Head returns some information about the object or an error if not found.
 	Head(key string) (Object, error)
 	// List returns a list of objects.
-	List(prefix, marker string, limit int64) ([]Object, error)
+	List(prefix, marker, delimiter string, limit int64) ([]Object, error)
 	// ListAll returns all the objects as an channel.
 	ListAll(prefix, marker string) (<-chan Object, error)
 

diff --git a/pkg/object/ks3.go b/pkg/object/ks3.go
@@ -27,7 +27,9 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/juicedata/juicefs/pkg/utils"
@@ -139,12 +141,13 @@ func (s *ks3) Delete(key string) error {
 	return err
 }
 
-func (s *ks3) List(prefix, marker string, limit int64) ([]Object, error) {
+func (s *ks3) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	param := s3.ListObjectsInput{
-		Bucket:  &s.bucket,
-		Prefix:  &prefix,
-		Marker:  &marker,
-		MaxKeys: &limit,
+		Bucket:    &s.bucket,
+		Prefix:    &prefix,
+		Marker:    &marker,
+		MaxKeys:   &limit,
+		Delimiter: &delimiter,
 	}
 	resp, err := s.s3.ListObjects(&param)
 	if err != nil {
@@ -156,6 +159,12 @@ func (s *ks3) List(prefix, marker string, limit int64) ([]Object, error) {
 		o := resp.Contents[i]
 		objs[i] = &obj{*o.Key, *o.Size, *o.LastModified, strings.HasSuffix(*o.Key, "/")}
 	}
+	if delimiter != "" {
+		for _, p := range resp.CommonPrefixes {
+			objs = append(objs, &obj{*p.Prefix, 0, time.Unix(0, 0), true})
+		}
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
+	}
 	return objs, nil
 }
 

diff --git a/pkg/object/mem.go b/pkg/object/mem.go
@@ -129,7 +129,10 @@ func (m *memStore) Delete(key string) error {
 	return nil
 }
 
-func (m *memStore) List(prefix, marker string, limit int64) ([]Object, error) {
+func (m *memStore) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
+	if delimiter != "" {
+		return nil, notSupportedDelimiter
+	}
 	m.Lock()
 	defer m.Unlock()
 

diff --git a/pkg/object/nos.go b/pkg/object/nos.go
@@ -27,6 +27,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"time"
 
@@ -130,12 +131,13 @@ func (s *nos) Delete(key string) error {
 	return s.client.DeleteObject(&param)
 }
 
-func (s *nos) List(prefix, marker string, limit int64) ([]Object, error) {
+func (s *nos) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	param := model.ListObjectsRequest{
-		Bucket:  s.bucket,
-		Prefix:  prefix,
-		Marker:  marker,
-		MaxKeys: int(limit),
+		Bucket:    s.bucket,
+		Prefix:    prefix,
+		Marker:    marker,
+		MaxKeys:   int(limit),
+		Delimiter: delimiter,
 	}
 	resp, err := s.client.ListObjects(&param)
 	if err != nil {
@@ -151,6 +153,12 @@ func (s *nos) List(prefix, marker string, limit int64) ([]Object, error) {
 		}
 		objs[i] = &obj{o.Key, o.Size, mtime, strings.HasSuffix(o.Key, "/")}
 	}
+	if delimiter != "" {
+		for _, p := range resp.CommonPrefixes {
+			objs = append(objs, &obj{p.Prefix, 0, time.Unix(0, 0), true})
+		}
+		sort.Slice(objs, func(i, j int) bool { return objs[i].Key() < objs[j].Key() })
+	}
 	return objs, nil
 }
 

diff --git a/pkg/object/object_storage.go b/pkg/object/object_storage.go
@@ -18,6 +18,7 @@ package object
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -103,6 +104,7 @@ type FileSystem interface {
 }
 
 var notSupported = utils.ENOTSUP
+var notSupportedDelimiter = errors.New("not supported delimiter")
 
 type DefaultObjectStorage struct{}
 
@@ -132,7 +134,7 @@ func (s DefaultObjectStorage) ListUploads(marker string) ([]*PendingPart, string
 	return nil, "", nil
 }
 
-func (s DefaultObjectStorage) List(prefix, marker string, limit int64) ([]Object, error) {
+func (s DefaultObjectStorage) List(prefix, marker, delimiter string, limit int64) ([]Object, error) {
 	return nil, notSupported
 }