You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jr...@apache.org on 2023/04/11 18:15:21 UTC
[beam] branch master updated: Go SDK: GCS FS supports rename, improved test coverage. Local FS supports Copy (#26065)
This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 78957dfe885 Go SDK: GCS FS supports rename, improved test coverage. Local FS supports Copy (#26065)
78957dfe885 is described below
commit 78957dfe8859d97caa2dc6835f3d57c8b2eaf782
Author: Jeremy Edwards <je...@gmail.com>
AuthorDate: Tue Apr 11 11:15:14 2023 -0700
Go SDK: GCS FS supports rename, improved test coverage. Local FS supports Copy (#26065)
---
CHANGES.md | 3 +-
sdks/go.mod | 9 +-
sdks/go.sum | 18 +-
sdks/go/pkg/beam/io/filesystem/gcs/gcs.go | 23 +++
sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go | 203 +++++++++++++++++++++
sdks/go/pkg/beam/io/filesystem/local/local.go | 17 ++
sdks/go/pkg/beam/io/filesystem/local/local_test.go | 61 +++++++
7 files changed, 329 insertions(+), 5 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 2ba06fb3514..b7a3df5b89c 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -58,7 +58,7 @@
## I/Os
-* Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* Added rename for GCS and copy for local filesystem (Go) ([#25779](https://github.com/apache/beam/issues/26064)).
## New Features / Improvements
@@ -80,7 +80,6 @@
* ([#X](https://github.com/apache/beam/issues/X)).
-
# [2.47.0] - Unreleased
## Highlights
diff --git a/sdks/go.mod b/sdks/go.mod
index f232237d32c..ec95ea518d5 100644
--- a/sdks/go.mod
+++ b/sdks/go.mod
@@ -29,7 +29,7 @@ require (
cloud.google.com/go/profiler v0.3.1
cloud.google.com/go/pubsub v1.30.0
cloud.google.com/go/spanner v1.44.0
- cloud.google.com/go/storage v1.29.0
+ cloud.google.com/go/storage v1.30.0
github.com/aws/aws-sdk-go-v2 v1.17.8
github.com/aws/aws-sdk-go-v2/config v1.18.19
github.com/aws/aws-sdk-go-v2/credentials v1.13.18
@@ -68,6 +68,8 @@ require (
require golang.org/x/exp v0.0.0-20230321023759-10a507213a29
+require github.com/fsouza/fake-gcs-server v1.44.1
+
require (
cloud.google.com/go v0.110.0 // indirect
cloud.google.com/go/compute v1.19.0 // indirect
@@ -106,14 +108,18 @@ require (
github.com/docker/go-units v0.5.0 // indirect
github.com/envoyproxy/go-control-plane v0.10.3 // indirect
github.com/envoyproxy/protoc-gen-validate v0.9.1 // indirect
+ github.com/felixge/httpsnoop v1.0.2 // indirect
github.com/goccy/go-json v0.9.11 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
github.com/golang/snappy v0.0.4 // indirect
github.com/google/flatbuffers v2.0.8+incompatible // indirect
github.com/google/pprof v0.0.0-20221103000818-d260c55eee4c // indirect
+ github.com/google/renameio/v2 v2.0.0 // indirect
github.com/googleapis/enterprise-certificate-proxy v0.2.3 // indirect
github.com/googleapis/gax-go/v2 v2.8.0 // indirect
+ github.com/gorilla/handlers v1.5.1 // indirect
+ github.com/gorilla/mux v1.8.0 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/klauspost/asmfmt v1.3.2 // indirect
@@ -133,6 +139,7 @@ require (
github.com/opencontainers/runc v1.1.5 // indirect
github.com/pierrec/lz4/v4 v4.1.15 // indirect
github.com/pkg/errors v0.9.1 // indirect
+ github.com/pkg/xattr v0.4.9 // indirect
github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 // indirect
github.com/shabbyrobe/gocovmerge v0.0.0-20180507124511-f6ea450bfb63 // indirect
github.com/sirupsen/logrus v1.9.0 // indirect
diff --git a/sdks/go.sum b/sdks/go.sum
index 29c40612a8a..e59939d8a54 100644
--- a/sdks/go.sum
+++ b/sdks/go.sum
@@ -54,8 +54,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-cloud.google.com/go/storage v1.29.0 h1:6weCgzRvMg7lzuUurI4697AqIRPU1SvzHhynwpW31jI=
-cloud.google.com/go/storage v1.29.0/go.mod h1:4puEjyTKnku6gfKoTfNOU/W+a9JyuVNxjpS5GBrB8h4=
+cloud.google.com/go/storage v1.30.0 h1:g1yrbxAWOrvg/594228pETWkOi00MLTrOWfh56veU5o=
+cloud.google.com/go/storage v1.30.0/go.mod h1:xAVretHSROm1BQX4IIsoVgJqw0LqOyX+I/O2GzRAzdE=
dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl2bqk7S3ta6S6u4k=
github.com/Azure/azure-storage-blob-go v0.14.0/go.mod h1:SMqIBi+SuiQH32bvyjngEewEeXoPfKMgWlBDaYf6fck=
@@ -208,10 +208,15 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
github.com/envoyproxy/protoc-gen-validate v0.6.7/go.mod h1:dyJXwwfPK2VSqiB9Klm1J6romD608Ba7Hij42vrOBCo=
github.com/envoyproxy/protoc-gen-validate v0.9.1 h1:PS7VIOgmSVhWUEeZwTe7z7zouA22Cr590PzXKbZHOVY=
github.com/envoyproxy/protoc-gen-validate v0.9.1/go.mod h1:OKNgG7TCp5pF4d6XftA0++PMirau2/yoOwVac3AbF2w=
+github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
+github.com/felixge/httpsnoop v1.0.2 h1:+nS9g82KMXccJ/wp0zyRW9ZBHFETmMGtkk+2CTTrW4o=
+github.com/felixge/httpsnoop v1.0.2/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
github.com/frankban/quicktest v1.2.2/go.mod h1:Qh/WofXFeiAFII1aEBu529AtJo6Zg2VHscnEsbBnJ20=
github.com/frankban/quicktest v1.11.3 h1:8sXhOn0uLys67V8EsXLc6eszDs8VXWxL3iRvebPhedY=
github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
+github.com/fsouza/fake-gcs-server v1.44.1 h1:G9tKIlnUNd6F9YIcrevq518nQk9FvOgUqHv7xsjGGRU=
+github.com/fsouza/fake-gcs-server v1.44.1/go.mod h1:HXlPE6QOJUoty9PTUlbT/w2iwVxH2HX6AZ459pAmCBE=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -304,6 +309,8 @@ github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hf
github.com/google/pprof v0.0.0-20221103000818-d260c55eee4c h1:lvddKcYTQ545ADhBujtIJmqQrZBDsGo7XIMbAQe/sNY=
github.com/google/pprof v0.0.0-20221103000818-d260c55eee4c/go.mod h1:dDKJzRmX4S37WGHujM7tX//fmj1uioxKzKxz3lo4HJo=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/renameio/v2 v2.0.0 h1:UifI23ZTGY8Tt29JbYFiuyIU3eX+RNFtUwefq9qAhxg=
+github.com/google/renameio/v2 v2.0.0/go.mod h1:BtmJXm5YlszgC+TD4HOEEUFgkJP3nLxehU6hfe7jRt4=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
@@ -314,6 +321,10 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
github.com/googleapis/gax-go/v2 v2.8.0 h1:UBtEZqx1bjXtOQ5BVTkuYghXrr3N4V123VKJK67vJZc=
github.com/googleapis/gax-go/v2 v2.8.0/go.mod h1:4orTrqY6hXxxaUL4LHIPl6lGo8vAE38/qKbhSAKP6QI=
+github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH4=
+github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
+github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0/go.mod h1:hgWBS7lorOAVIJEQMi4ZsPv9hVvWI6+ch50m39Pf2Ks=
github.com/hashicorp/go-uuid v0.0.0-20180228145832-27454136f036/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
@@ -395,6 +406,8 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
+github.com/pkg/xattr v0.4.9 h1:5883YPCtkSd8LFbs13nXplj9g9tlrwoJRjgpgMu1/fE=
+github.com/pkg/xattr v0.4.9/go.mod h1:di8WF84zAKk8jzR1UBTEWh9AUlIZZ7M/JNt8e9B6ktU=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -633,6 +646,7 @@ golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220408201424-a24fb2fb8a0f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
diff --git a/sdks/go/pkg/beam/io/filesystem/gcs/gcs.go b/sdks/go/pkg/beam/io/filesystem/gcs/gcs.go
index 1dc2f85275d..34fe2cf16fe 100644
--- a/sdks/go/pkg/beam/io/filesystem/gcs/gcs.go
+++ b/sdks/go/pkg/beam/io/filesystem/gcs/gcs.go
@@ -166,8 +166,31 @@ func (f *fs) Copy(ctx context.Context, srcpath, dstpath string) error {
return err
}
+// Rename the old path to the new path.
+func (f *fs) Rename(ctx context.Context, srcpath, dstpath string) error {
+ bucket, src, err := gcsx.ParseObject(srcpath)
+ if err != nil {
+ return err
+ }
+ srcobj := f.client.Bucket(bucket).Object(src)
+
+ bucket, dst, err := gcsx.ParseObject(dstpath)
+ if err != nil {
+ return err
+ }
+ dstobj := f.client.Bucket(bucket).Object(dst)
+
+ cp := dstobj.CopierFrom(srcobj)
+ _, err = cp.Run(ctx)
+ if err != nil {
+ return err
+ }
+ return srcobj.Delete(ctx)
+}
+
// Compile time check for interface implementations.
var (
_ filesystem.Remover = ((*fs)(nil))
_ filesystem.Copier = ((*fs)(nil))
+ _ filesystem.Renamer = ((*fs)(nil))
)
diff --git a/sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go b/sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go
index c1c2cc16c5d..66e2d67813a 100644
--- a/sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go
+++ b/sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go
@@ -17,9 +17,13 @@ package gcs
import (
"context"
+ "io"
+ "sort"
"testing"
"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+ "github.com/fsouza/fake-gcs-server/fakestorage"
+ "github.com/google/go-cmp/cmp"
)
func TestGCS_FilesystemNew(t *testing.T) {
@@ -36,3 +40,202 @@ func TestGCS_FilesystemNew(t *testing.T) {
t.Errorf("c.Close() = %v, want nil", err)
}
}
+
+func TestGCS_direct(t *testing.T) {
+ ctx := context.Background()
+ dirPath := "gs://beamgogcsfilesystemtest"
+ filePath := dirPath + "/file.txt"
+
+ server := createFakeGCSServer(t)
+ c := &fs{client: server.Client()}
+
+ wc, err := c.OpenWrite(ctx, filePath)
+ if err != nil {
+ t.Fatalf("OpenWrite(ctx, %q) == %v, want nil", filePath, err)
+ }
+ t.Cleanup(func() {
+ wc.Close()
+ })
+ data := []byte("a meaningless sequence of test words")
+ n, err := wc.Write(data)
+ if got, want := n, len(data); err != nil || got != want {
+ t.Fatalf("Write(data) = %v,%v, want %v, nil", got, err, want)
+ }
+ wc.Close()
+
+ listGlob := dirPath + "/*.txt"
+ files, err := c.List(ctx, listGlob)
+ if err != nil {
+ t.Fatalf("List(%q) = %v, want nil", listGlob, err)
+ }
+
+ if len(files) != 1 || (len(files) == 1 && files[0] != filePath) {
+ t.Errorf("List(%v) = %v, want []string{%v}", listGlob, files, filePath)
+ }
+
+ size, err := c.Size(ctx, filePath)
+ if got, want := size, int64(len(data)); err != nil || got != want {
+ t.Errorf("Size(%q) = %v, %v, want %v, nil", filePath, got, err, want)
+ }
+
+ rc, err := c.OpenRead(ctx, filePath)
+ if err != nil {
+ t.Fatalf("OpenRead(ctx, %q) == %v, want nil", filePath, err)
+ }
+ t.Cleanup(func() {
+ c.Close()
+ })
+
+ buf, err := io.ReadAll(rc)
+ if got, want := n, len(buf); err != nil || got != want {
+ t.Fatalf("ReadAll() = %v,%v, want %v, nil", got, err, want)
+ }
+ if got, want := string(data), string(buf); got != want {
+ t.Errorf("ReadAll() = %v, want %v", got, want)
+ }
+}
+
+func TestLocal_util(t *testing.T) {
+ ctx := context.Background()
+ dirPath := "gs://beamgogcsfilesystemtest"
+ filePath := dirPath + "/file.txt"
+
+ server := createFakeGCSServer(t)
+ c := &fs{client: server.Client()}
+
+ data := []byte("a meaningless sequence of test words")
+ if err := filesystem.Write(ctx, c, filePath, data); err != nil {
+ t.Fatalf("filesystem.Write(ctx, %q) == %v, want nil", filePath, err)
+ }
+
+ gotData, err := filesystem.Read(ctx, c, filePath)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+}
+func TestLocal_listNoMatches(t *testing.T) {
+ ctx := context.Background()
+ server := createFakeGCSServer(t)
+ c := &fs{client: server.Client()}
+
+ glob := "gs://beamgogcsfilesystemtest/foo*"
+
+ got, err := c.List(ctx, glob)
+ if err != nil {
+ t.Fatalf("List(%q) error = %v, want nil", glob, err)
+ }
+
+ want := []string(nil)
+ if !cmp.Equal(got, want) {
+ t.Errorf("List(%q) = %v, want %v", glob, got, want)
+ }
+}
+
+func TestGCS_rename(t *testing.T) {
+ ctx := context.Background()
+ dirPath := "gs://beamgogcsfilesystemtest"
+ filePath1 := dirPath + "/file1.txt"
+ filePath2 := dirPath + "/file2.txt"
+
+ server := createFakeGCSServer(t)
+ c := &fs{client: server.Client()}
+
+ data := []byte("a meaningless sequence of test words")
+ if err := filesystem.Write(ctx, c, filePath1, data); err != nil {
+ t.Fatalf("filesystem.Write(ctx, %q) == %v, want nil", filePath1, err)
+ }
+
+ if err := filesystem.Rename(ctx, c, filePath1, filePath2); err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath2, err)
+ }
+
+ gotData, err := filesystem.Read(ctx, c, filePath2)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath2, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+
+ // Check that the old file doesn't exist.
+ listGlob := dirPath + "/*.txt"
+ files, err := c.List(ctx, listGlob)
+ if err != nil {
+ t.Fatalf("List(%q) = %v, want nil", listGlob, err)
+ }
+
+ if len(files) != 1 || (len(files) == 1 && files[0] != filePath2) {
+ t.Errorf("List(%v) = %v, want []string{%v}", listGlob, files, filePath2)
+ }
+}
+
+func TestGCS_copy(t *testing.T) {
+ ctx := context.Background()
+ dirPath := "gs://beamgogcsfilesystemtest"
+ filePath1 := dirPath + "/file1.txt"
+ filePath2 := dirPath + "/file2.txt"
+
+ server := createFakeGCSServer(t)
+ c := &fs{client: server.Client()}
+
+ data := []byte("a meaningless sequence of test words")
+ if err := filesystem.Write(ctx, c, filePath1, data); err != nil {
+ t.Fatalf("filesystem.Write(ctx, %q) == %v, want nil", filePath1, err)
+ }
+
+ if err := filesystem.Copy(ctx, c, filePath1, filePath2); err != nil {
+ t.Fatalf("filesystem.Copy(ctx, %q) == %v, want nil", filePath2, err)
+ }
+
+ gotData, err := filesystem.Read(ctx, c, filePath2)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath2, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+
+ // Reread old data to make sure that the file still exists.
+ gotData, err = filesystem.Read(ctx, c, filePath1)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath1, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+
+ listGlob := dirPath + "/*.txt"
+ files, err := c.List(ctx, listGlob)
+ if err != nil {
+ t.Fatalf("List(%q) = %v, want nil", listGlob, err)
+ }
+
+ // Sort the strings to ensure stable output.
+ sort.Strings(files)
+
+ if files[0] != filePath1 {
+ t.Errorf("List(%s) = %v, want []string{%s, %s}", listGlob, files, filePath1, filePath2)
+ }
+ if files[1] != filePath2 {
+ t.Errorf("List(%s) = %v, want []string{%s, %s}", listGlob, files, filePath1, filePath2)
+ }
+}
+
+func createFakeGCSServer(tb testing.TB) *fakestorage.Server {
+ tb.Helper()
+
+ server := fakestorage.NewServer([]fakestorage.Object{
+ {
+ ObjectAttrs: fakestorage.ObjectAttrs{
+ BucketName: "beamgogcsfilesystemtest",
+ Name: "stub",
+ },
+ Content: []byte(""),
+ },
+ })
+ tb.Cleanup(server.Stop)
+ return server
+}
diff --git a/sdks/go/pkg/beam/io/filesystem/local/local.go b/sdks/go/pkg/beam/io/filesystem/local/local.go
index 0cb0060451c..0cc58c4fb32 100644
--- a/sdks/go/pkg/beam/io/filesystem/local/local.go
+++ b/sdks/go/pkg/beam/io/filesystem/local/local.go
@@ -77,8 +77,25 @@ func (f *fs) Rename(_ context.Context, oldpath, newpath string) error {
return os.Rename(oldpath, newpath)
}
+// Copy copies from oldpath to the newpath.
+func (f *fs) Copy(_ context.Context, oldpath, newpath string) error {
+ srcFile, err := os.Open(oldpath)
+ if err != nil {
+ return err
+ }
+ defer srcFile.Close()
+ destFile, err := os.Create(newpath)
+ if err != nil {
+ return err
+ }
+ defer destFile.Close()
+ _, err = io.Copy(destFile, srcFile)
+ return err
+}
+
// Compile time check for interface implementations.
var (
+ _ filesystem.Copier = ((*fs)(nil))
_ filesystem.Remover = ((*fs)(nil))
_ filesystem.Renamer = ((*fs)(nil))
)
diff --git a/sdks/go/pkg/beam/io/filesystem/local/local_test.go b/sdks/go/pkg/beam/io/filesystem/local/local_test.go
index f5ef2000df7..c8be61a3371 100644
--- a/sdks/go/pkg/beam/io/filesystem/local/local_test.go
+++ b/sdks/go/pkg/beam/io/filesystem/local/local_test.go
@@ -20,6 +20,7 @@ import (
"io"
"os"
"path/filepath"
+ "sort"
"testing"
"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
@@ -197,3 +198,63 @@ func TestLocal_rename(t *testing.T) {
t.Errorf("List(%v) = %v, want []string{%v}", listGlob, files, filePath2)
}
}
+
+func TestLocal_copy(t *testing.T) {
+ ctx := context.Background()
+ tempDir := t.TempDir()
+ dirPath := filepath.Join(tempDir, "beamgolocalfilesystemtest")
+ filePath1 := filepath.Join(dirPath, "file1.txt")
+ filePath2 := filepath.Join(dirPath, "file2.txt")
+ t.Cleanup(func() {
+ os.RemoveAll(tempDir)
+ })
+
+ c, err := filesystem.New(ctx, dirPath)
+ if err != nil {
+ t.Fatalf("filesystem.New(ctx, %q) == %v, want nil", dirPath, err)
+ }
+ t.Cleanup(func() {
+ c.Close()
+ })
+ data := []byte("a meaningless sequence of test words")
+ if err := filesystem.Write(ctx, c, filePath1, data); err != nil {
+ t.Fatalf("filesystem.Write(ctx, %q) == %v, want nil", filePath1, err)
+ }
+
+ if err := filesystem.Copy(ctx, c, filePath1, filePath2); err != nil {
+ t.Fatalf("filesystem.Copy(ctx, %q) == %v, want nil", filePath2, err)
+ }
+
+ gotData, err := filesystem.Read(ctx, c, filePath2)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath2, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+
+ // Reread old data to make sure that the file still exists.
+ gotData, err = filesystem.Read(ctx, c, filePath1)
+ if err != nil {
+ t.Fatalf("filesystem.Read(ctx, %q) == %v, want nil", filePath1, err)
+ }
+ if got, want := string(data), string(gotData); got != want {
+ t.Errorf("filesystem.Read() = %v, want %v", got, want)
+ }
+
+ listGlob := filepath.Join(dirPath, "*.txt")
+ files, err := c.List(ctx, listGlob)
+ if err != nil {
+ t.Fatalf("List(%q) = %v, want nil", listGlob, err)
+ }
+
+ // Sort the strings to ensure stable output.
+ sort.Strings(files)
+
+ if files[0] != filePath1 {
+ t.Errorf("List(%s) = %v, want []string{%s, %s}", listGlob, files, filePath1, filePath2)
+ }
+ if files[1] != filePath2 {
+ t.Errorf("List(%s) = %v, want []string{%s, %s}", listGlob, files, filePath1, filePath2)
+ }
+}