Add minio fs unit tests (#43)

Signed-off-by: sunby <[email protected]>

.github/workflows/ci.yaml

@@ -22,10 +22,19 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Install dependencies
         run: cd go && go mod download
 
+      - name: Setup minio
+        run: |
+          docker run -d -p 9000:9000 --name minio \
+                     -e "MINIO_ACCESS_KEY=minioadmin" \
+                     -e "MINIO_SECRET_KEY=minioadmin" \
+                     -v /tmp/data:/data \
+                     -v /tmp/config:/root/.minio \
+                     minio/minio server /data
+
       - name: Run tests
         run: cd go && go test -v ./...

go/go.mod

@@ -1,56 +1,59 @@
 module github.com/milvus-io/milvus-storage-format
 
-go 1.18
+go 1.19
 
 require (
 	github.com/apache/arrow/go/v12 v12.0.0-20230223012627-e0e740bd7a24
 	github.com/bits-and-blooms/bitset v1.5.0
 	github.com/google/uuid v1.3.0
-	github.com/stretchr/testify v1.8.2
+	github.com/minio/minio-go/v7 v7.0.61
+	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.24.0
-	google.golang.org/protobuf v1.28.1
+	google.golang.org/grpc v1.57.0
+	google.golang.org/protobuf v1.31.0
 )
 
 require (
 	github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
-	github.com/apache/thrift v0.16.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/apache/thrift v0.18.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
-	github.com/goccy/go-json v0.9.11 // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/goccy/go-json v0.10.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/flatbuffers v2.0.8+incompatible // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/asmfmt v1.3.2 // indirect
 	github.com/klauspost/compress v1.16.7 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
-	github.com/kr/text v0.2.0 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
-	github.com/minio/minio-go/v7 v7.0.61 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/pierrec/lz4/v4 v4.1.15 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.18 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	github.com/rs/xid v1.5.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
-	go.uber.org/atomic v1.7.0 // indirect
-	go.uber.org/multierr v1.6.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/goleak v1.2.1 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.11.0 // indirect
-	golang.org/x/exp v0.0.0-20220827204233-334a2380cb91 // indirect
-	golang.org/x/mod v0.8.0 // indirect
+	golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691 // indirect
+	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/net v0.12.0 // indirect
-	golang.org/x/sync v0.1.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
 	golang.org/x/sys v0.10.0 // indirect
 	golang.org/x/text v0.11.0 // indirect
-	golang.org/x/tools v0.6.0 // indirect
-	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect
-	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
-	google.golang.org/grpc v1.49.0 // indirect
+	golang.org/x/tools v0.11.0 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230726155614-23370e0ffb3e // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go/io/fs/file/minio_file.go

@@ -22,7 +22,7 @@ func (f *MinioFile) Write(b []byte) (int, error) {
 }
 
 func (f *MinioFile) Close() error {
-	if f.writer == nil {
+	if len(f.writer.b) == 0 {
 		return nil
 	}
 	_, err := f.client.PutObject(context.TODO(), f.bucketName, f.fileName, bytes.NewReader(f.writer.b), int64(len(f.writer.b)), minio.PutObjectOptions{})
@@ -51,6 +51,7 @@ func NewMinioFile(client *minio.Client, fileName string, bucketName string) (*Mi
 
 	return &MinioFile{
 		Object:     object,
+		writer:     NewMemoryFile(nil),
 		client:     client,
 		fileName:   fileName,
 		bucketName: bucketName,

go/io/fs/fs.go

@@ -11,6 +11,7 @@ type Fs interface {
 	CreateDir(path string) error
 	List(path string) ([]FileEntry, error)
 	ReadFile(path string) ([]byte, error)
+	Exist(path string) (bool, error)
 }
 type FileEntry struct {
 	Path string

go/io/fs/local_fs.go

@@ -61,6 +61,10 @@ func (l *LocalFS) ReadFile(path string) ([]byte, error) {
 	return os.ReadFile(path)
 }
 
+func (l *LocalFS) Exist(path string) (bool, error) {
+	panic("not implemented")
+}
+
 func NewLocalFs() *LocalFS {
 	return &LocalFS{}
 }

go/io/fs/memory_fs.go

@@ -44,6 +44,10 @@ func (m *MemoryFs) ReadFile(path string) ([]byte, error) {
 	panic("implement me")
 }
 
+func (m *MemoryFs) Exist(path string) (bool, error) {
+	panic("not implemented")
+}
+
 func NewMemoryFs() *MemoryFs {
 	return &MemoryFs{
 		files: make(map[string]*file.MemoryFile),

go/io/fs/minio_fs.go

@@ -3,6 +3,7 @@ package fs
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/url"
 
 	"github.com/milvus-io/milvus-storage-format/common/log"
@@ -43,7 +44,7 @@ func (fs *MinioFs) CreateDir(path string) error {
 
 func (fs *MinioFs) List(path string) ([]FileEntry, error) {
 	ret := make([]FileEntry, 0)
-	for objInfo := range fs.client.ListObjects(context.TODO(), fs.bucketName, minio.ListObjectsOptions{Prefix: path}) {
+	for objInfo := range fs.client.ListObjects(context.TODO(), fs.bucketName, minio.ListObjectsOptions{Prefix: path, Recursive: false}) {
 		if objInfo.Err != nil {
 			log.Warn("list object error", zap.Error(objInfo.Err))
 			return nil, objInfo.Err
@@ -64,9 +65,9 @@ func (fs *MinioFs) ReadFile(path string) ([]byte, error) {
 		return nil, err
 	}
 
-	buf := make([]byte, 0, stat.Size)
+	buf := make([]byte, stat.Size)
 	n, err := obj.Read(buf)
-	if err != nil {
+	if err != nil && err != io.EOF {
 		return nil, err
 	}
 	if n != int(stat.Size) {
@@ -75,6 +76,18 @@ func (fs *MinioFs) ReadFile(path string) ([]byte, error) {
 	return buf, nil
 }
 
+func (fs *MinioFs) Exist(path string) (bool, error) {
+	_, err := fs.client.StatObject(context.TODO(), fs.bucketName, path, minio.StatObjectOptions{})
+	if err != nil {
+		resp := minio.ToErrorResponse(err)
+		if resp.Code == "NoSuchKey" {
+			return false, nil
+		}
+		return false, err
+	}
+	return true, nil
+}
+
 // uri should be s3://accessKey:secretAceessKey@endpoint/bucket/
 func NewMinioFs(uri *url.URL) (*MinioFs, error) {
 	accessKey := uri.User.Username()
@@ -90,19 +103,23 @@ func NewMinioFs(uri *url.URL) (*MinioFs, error) {
 		return nil, err
 	}
 
-	exist, err := cli.BucketExists(context.TODO(), uri.Path)
+	bucket := uri.Path
+	if bucket[0] == '/' {
+		bucket = bucket[1:]
+	}
+	exist, err := cli.BucketExists(context.TODO(), bucket)
 	if err != nil {
 		return nil, err
 	}
 
 	if !exist {
-		if err = cli.MakeBucket(context.TODO(), uri.Path, minio.MakeBucketOptions{}); err != nil {
+		if err = cli.MakeBucket(context.TODO(), bucket, minio.MakeBucketOptions{}); err != nil {
 			return nil, err
 		}
 	}
 
 	return &MinioFs{
 		client:     cli,
-		bucketName: uri.Path,
+		bucketName: bucket,
 	}, nil
 }

go/io/fs/minio_fs_test.go

@@ -0,0 +1,126 @@
+package fs_test
+
+import (
+	"io"
+	"testing"
+
+	"github.com/milvus-io/milvus-storage-format/io/fs"
+	"github.com/minio/minio-go/v7"
+	"github.com/stretchr/testify/suite"
+)
+
+type MinioFsTestSuite struct {
+	suite.Suite
+	fs     fs.Fs
+	client *minio.Client
+}
+
+func (suite *MinioFsTestSuite) SetupSuite() {
+	fs, err := fs.BuildFileSystem("s3://minioadmin:minioadmin@localhost:9000/default")
+	suite.NoError(err)
+	suite.fs = fs
+}
+
+func (suite *MinioFsTestSuite) TestMinioOpenFile() {
+	file, err := suite.fs.OpenFile("a")
+	suite.NoError(err)
+	n, err := file.Write([]byte{1})
+	suite.NoError(err)
+	suite.Equal(1, n)
+	suite.NoError(file.Close())
+
+	file, err = suite.fs.OpenFile("a")
+	suite.NoError(err)
+	buf := make([]byte, 10)
+	n, err = file.Read(buf)
+	suite.Equal(io.EOF, err)
+	suite.Equal(1, n)
+	suite.ElementsMatch(buf[:n], []byte{1})
+}
+
+func (suite *MinioFsTestSuite) TestMinioRename() {
+	file, err := suite.fs.OpenFile("a")
+	suite.NoError(err)
+	n, err := file.Write([]byte{1})
+	suite.NoError(err)
+	suite.Equal(1, n)
+	suite.NoError(file.Close())
+
+	err = suite.fs.Rename("a", "b")
+	suite.NoError(err)
+
+	file, err = suite.fs.OpenFile("b")
+	suite.NoError(err)
+	buf := make([]byte, 10)
+	n, err = file.Read(buf)
+	suite.Equal(io.EOF, err)
+	suite.Equal(1, n)
+	suite.ElementsMatch(buf[:n], []byte{1})
+}
+
+func (suite *MinioFsTestSuite) TestMinioFsDeleteFile() {
+	file, err := suite.fs.OpenFile("a")
+	suite.NoError(err)
+	n, err := file.Write([]byte{1})
+	suite.NoError(err)
+	suite.Equal(1, n)
+	suite.NoError(file.Close())
+
+	err = suite.fs.DeleteFile("a")
+	suite.NoError(err)
+
+	exist, err := suite.fs.Exist("a")
+	suite.NoError(err)
+	suite.False(exist)
+}
+
+func (suite *MinioFsTestSuite) TestMinioFsList() {
+	file, err := suite.fs.OpenFile("a/b")
+	suite.NoError(err)
+	_, err = file.Write([]byte{1})
+	suite.NoError(err)
+	suite.NoError(file.Close())
+	file, err = suite.fs.OpenFile("a/b/c")
+	suite.NoError(err)
+	_, err = file.Write([]byte{1})
+	suite.NoError(err)
+	suite.NoError(file.Close())
+
+	entries, err := suite.fs.List("a/")
+	suite.NoError(err)
+	suite.EqualValues([]fs.FileEntry{{Path: "a/b"}}, entries)
+}
+
+func (suite *MinioFsTestSuite) TestMinioFsReadFile() {
+	file, err := suite.fs.OpenFile("a")
+	suite.NoError(err)
+	n, err := file.Write([]byte{1})
+	suite.NoError(err)
+	suite.Equal(1, n)
+	suite.NoError(file.Close())
+
+	content, err := suite.fs.ReadFile("a")
+	suite.NoError(err)
+	suite.EqualValues([]byte{1}, content)
+}
+
+func (suite *MinioFsTestSuite) TestMinioFsExist() {
+	exist, err := suite.fs.Exist("nonexist")
+	suite.NoError(err)
+	suite.False(exist)
+
+	file, err := suite.fs.OpenFile("exist")
+	suite.NoError(err)
+	n, err := file.Write([]byte{1})
+	suite.NoError(err)
+	suite.Equal(1, n)
+	suite.NoError(file.Close())
+
+	exist, err = suite.fs.Exist("exist")
+	suite.NoError(err)
+	suite.True(exist)
+}
+
+func TestMinioFsSuite(t *testing.T) {
+	suite.Run(t, &MinioFsTestSuite{})
+}