From a264a6500d055effe535f5bc9f1b2f314d2df4cc Mon Sep 17 00:00:00 2001 From: badalprasadsingh Date: Sat, 29 Aug 2026 04:49:37 +0530 Subject: [PATCH 1/3] feat: split into per-cloud backend packages Signed-off-by: badalprasadsingh --- catalog/hadoop/io.go | 4 +- internal/schemes/schemes.go | 42 +++++ io/gocloud/{ => azure}/azure.go | 29 ++-- .../{ => azure}/azure_integration_test.go | 4 +- io/gocloud/{ => azure}/azure_test.go | 80 ++++++++- io/gocloud/azure/register.go | 47 ++++++ io/gocloud/azure/register_test.go | 52 ++++++ io/gocloud/{ => blobfs}/blob.go | 96 ++++++----- io/gocloud/{ => blobfs}/blob_test.go | 152 ++++-------------- io/gocloud/{ => blobfs}/utils.go | 4 +- io/gocloud/compat_test.go | 100 ++++++++++++ io/gocloud/{ => gcs}/gcs.go | 2 +- io/gocloud/{ => gcs}/gcs_integration_test.go | 4 +- io/gocloud/{ => gcs}/gcs_test.go | 2 +- io/gocloud/gcs/register.go | 47 ++++++ io/gocloud/gcs/register_test.go | 52 ++++++ io/gocloud/gocloud.go | 66 ++++++++ io/gocloud/isolation_test.go | 99 ++++++++++++ io/gocloud/register.go | 86 ---------- io/gocloud/s3/register.go | 47 ++++++ io/gocloud/s3/register_test.go | 52 ++++++ io/gocloud/{ => s3}/s3.go | 2 +- io/gocloud/{ => s3}/s3_integration_test.go | 4 +- io/gocloud/{ => s3}/s3_test.go | 2 +- io/io.go | 10 +- io/registry.go | 15 +- website/src/configuration.md | 19 ++- website/src/feature-status.md | 2 +- website/src/getting-started.md | 2 +- 29 files changed, 823 insertions(+), 300 deletions(-) create mode 100644 internal/schemes/schemes.go rename io/gocloud/{ => azure}/azure.go (89%) rename io/gocloud/{ => azure}/azure_integration_test.go (98%) rename io/gocloud/{ => azure}/azure_test.go (79%) create mode 100644 io/gocloud/azure/register.go create mode 100644 io/gocloud/azure/register_test.go rename io/gocloud/{ => blobfs}/blob.go (85%) rename io/gocloud/{ => blobfs}/blob_test.go (85%) rename io/gocloud/{ => blobfs}/utils.go (92%) create mode 100644 io/gocloud/compat_test.go rename io/gocloud/{ => gcs}/gcs.go (99%) rename io/gocloud/{ => gcs}/gcs_integration_test.go (98%) rename io/gocloud/{ => gcs}/gcs_test.go (99%) create mode 100644 io/gocloud/gcs/register.go create mode 100644 io/gocloud/gcs/register_test.go create mode 100644 io/gocloud/gocloud.go create mode 100644 io/gocloud/isolation_test.go delete mode 100644 io/gocloud/register.go create mode 100644 io/gocloud/s3/register.go create mode 100644 io/gocloud/s3/register_test.go rename io/gocloud/{ => s3}/s3.go (99%) rename io/gocloud/{ => s3}/s3_integration_test.go (98%) rename io/gocloud/{ => s3}/s3_test.go (99%) diff --git a/catalog/hadoop/io.go b/catalog/hadoop/io.go index cee666997..58b8e52d3 100644 --- a/catalog/hadoop/io.go +++ b/catalog/hadoop/io.go @@ -19,7 +19,7 @@ package hadoop import ( icebergio "github.com/apache/iceberg-go/io" - "github.com/apache/iceberg-go/io/gocloud" + "github.com/apache/iceberg-go/io/gocloud/blobfs" ) // HadoopCatalogFS represents all the interfaces that a filesystem implementation @@ -39,4 +39,4 @@ type HadoopCatalogFS interface { var _ HadoopCatalogFS = (*icebergio.LocalFS)(nil) // BlobFileIO can be used to implement a Hadoop catalog with a blob storage bucket. -var _ HadoopCatalogFS = (*gocloud.BlobFileIO)(nil) +var _ HadoopCatalogFS = (*blobfs.FileIO)(nil) diff --git a/internal/schemes/schemes.go b/internal/schemes/schemes.go new file mode 100644 index 000000000..c3810b068 --- /dev/null +++ b/internal/schemes/schemes.go @@ -0,0 +1,42 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package schemes + +import "slices" + +var ( + S3 = []string{"s3", "s3a", "s3n", "oss"} + GCS = []string{"gs"} + Azure = []string{"abfs", "abfss", "wasb", "wasbs"} +) + +var byBackend = map[string][]string{ + "s3": S3, + "gcs": GCS, + "azure": Azure, +} + +func BackendFor(scheme string) string { + for backend, list := range byBackend { + if slices.Contains(list, scheme) { + return backend + } + } + + return "" +} diff --git a/io/gocloud/azure.go b/io/gocloud/azure/azure.go similarity index 89% rename from io/gocloud/azure.go rename to io/gocloud/azure/azure.go index 2b1ee3925..71bcad919 100644 --- a/io/gocloud/azure.go +++ b/io/gocloud/azure/azure.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package azure import ( "context" @@ -29,6 +29,7 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/io/gocloud/blobfs" "gocloud.dev/blob" "gocloud.dev/blob/azureblob" ) @@ -108,8 +109,8 @@ func newAdlsLocation(adlsURI *url.URL) (*adlsLocation, error) { // Construct a Azure bucket from a URL func createAzureBucket(ctx context.Context, parsed *url.URL, props map[string]string) (*blob.Bucket, error) { - adlsSasTokens := propertiesWithPrefix(props, io.ADLSSasTokenPrefix) - adlsConnectionStrings := propertiesWithPrefix(props, io.ADLSConnectionStringPrefix) + adlsSasTokens := blobfs.PropertiesWithPrefix(props, io.ADLSSasTokenPrefix) + adlsConnectionStrings := blobfs.PropertiesWithPrefix(props, io.ADLSConnectionStringPrefix) // Construct the client location, err := newAdlsLocation(parsed) @@ -209,40 +210,34 @@ func adlsAuthority(parsed *url.URL) string { return parsed.User.Username() + "@" + parsed.Hostname() } -func adlsObjectLocationExtractor(parsedURL *url.URL) objectLocationExtractor { +func adlsObjectLocationExtractor(parsedURL *url.URL) blobfs.ObjectLocationExtractor { expectedAuthority := adlsAuthority(parsedURL) - return func(location string) (objectLocation, error) { + return func(location string) (blobfs.ObjectLocation, error) { matches := adlsURIPattern.FindStringSubmatch(location) if len(matches) < 4 { - return objectLocation{}, fmt.Errorf("invalid ADLS location: %s", location) + return blobfs.ObjectLocation{}, fmt.Errorf("invalid ADLS location: %s", location) } authority := matches[2] if authority != expectedAuthority { - return objectLocation{}, fmt.Errorf("%w: URI authority %q does not match configured authority %q", - ErrUnsupportedObjectAuthority, + return blobfs.ObjectLocation{}, fmt.Errorf("%w: URI authority %q does not match configured authority %q", + blobfs.ErrUnsupportedObjectAuthority, authority, expectedAuthority) } uriPath := matches[3] if uriPath != "" && !strings.HasPrefix(uriPath, "/") { - return objectLocation{}, fmt.Errorf("URI authority %q must be followed by an object path: %s", + return blobfs.ObjectLocation{}, fmt.Errorf("URI authority %q must be followed by an object path: %s", authority, location) } key := strings.TrimPrefix(uriPath, "/") - parsed := objectLocation{ - scheme: matches[1], - authority: authority, - key: key, - uriPrefix: matches[1] + "://" + authority + "/", - hasAuthority: true, - } + parsed := blobfs.NewObjectLocation(matches[1], authority, key) if key == "" { - return parsed, fmt.Errorf("%w: %s", ErrEmptyObjectKey, location) + return parsed, fmt.Errorf("%w: %s", blobfs.ErrEmptyObjectKey, location) } return parsed, nil diff --git a/io/gocloud/azure_integration_test.go b/io/gocloud/azure/azure_integration_test.go similarity index 98% rename from io/gocloud/azure_integration_test.go rename to io/gocloud/azure/azure_integration_test.go index cd6998400..bf4da9ef7 100644 --- a/io/gocloud/azure_integration_test.go +++ b/io/gocloud/azure/azure_integration_test.go @@ -17,7 +17,7 @@ //go:build integration -package gocloud_test +package azure_test import ( "context" @@ -30,7 +30,7 @@ import ( "github.com/apache/iceberg-go/catalog" sqlcat "github.com/apache/iceberg-go/catalog/sql" "github.com/apache/iceberg-go/io" - _ "github.com/apache/iceberg-go/io/gocloud" + _ "github.com/apache/iceberg-go/io/gocloud/azure" "github.com/stretchr/testify/suite" "github.com/uptrace/bun/driver/sqliteshim" "gocloud.dev/blob/azureblob" diff --git a/io/gocloud/azure_test.go b/io/gocloud/azure/azure_test.go similarity index 79% rename from io/gocloud/azure_test.go rename to io/gocloud/azure/azure_test.go index 5d26960a4..e98c8f197 100644 --- a/io/gocloud/azure_test.go +++ b/io/gocloud/azure/azure_test.go @@ -15,15 +15,19 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package azure import ( "context" + "io/fs" "net/url" "testing" + "github.com/apache/iceberg-go/io/gocloud/blobfs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gocloud.dev/blob" + "gocloud.dev/blob/memblob" ) func TestCreateAzureBucketDefaultCredentialCalled(t *testing.T) { @@ -207,14 +211,14 @@ func TestAdlsKeyExtractor(t *testing.T) { name: "URI with no path", input: "abfs://container@account.dfs.core.windows.net", expectedErr: "object key is empty", - wantErrIs: ErrEmptyObjectKey, + wantErrIs: blobfs.ErrEmptyObjectKey, shouldError: true, }, { name: "URI with empty path", input: "abfs://container@account.dfs.core.windows.net/", expectedErr: "object key is empty", - wantErrIs: ErrEmptyObjectKey, + wantErrIs: blobfs.ErrEmptyObjectKey, shouldError: true, }, { @@ -227,7 +231,7 @@ func TestAdlsKeyExtractor(t *testing.T) { name: "URI with different container", input: "abfs://other@account.dfs.core.windows.net/path/to/file.parquet", expectedErr: "does not match configured authority", - wantErrIs: ErrUnsupportedObjectAuthority, + wantErrIs: blobfs.ErrUnsupportedObjectAuthority, shouldError: true, }, { @@ -247,7 +251,7 @@ func TestAdlsKeyExtractor(t *testing.T) { parsed, err := url.Parse(root) require.NoError(t, err) - extractor := keyExtractorFromObjectLocation(adlsObjectLocationExtractor(parsed)) + extractor := blobfs.KeyExtractorFromObjectLocation(adlsObjectLocationExtractor(parsed)) key, err := extractor(test.input) if test.shouldError { @@ -265,3 +269,69 @@ func TestAdlsKeyExtractor(t *testing.T) { }) } } + +func testADLSBlobFileIO(t *testing.T, ctx context.Context, root string, bucket *blob.Bucket) *blobfs.FileIO { + t.Helper() + + parsed, err := url.Parse(root) + require.NoError(t, err) + + return blobfs.New(ctx, bucket, adlsObjectLocationExtractor(parsed)) +} + +func TestBlobFileIOWalkDirRejectsWrongAzureAuthority(t *testing.T) { + ctx := context.Background() + + bucket := memblob.OpenBucket(nil) + defer bucket.Close() + + bfs := testADLSBlobFileIO(t, ctx, "abfs://container@account.dfs.core.windows.net/", bucket) + + err := bfs.WalkDir("abfs://other@account.dfs.core.windows.net/data", func(string, fs.DirEntry, error) error { + t.Fatal("WalkDir callback should not be called") + + return nil + }) + require.ErrorContains(t, err, "does not match configured authority") + require.ErrorIs(t, err, blobfs.ErrUnsupportedObjectAuthority) +} + +func TestBlobFileIOWalkDirAzureURI(t *testing.T) { + ctx := context.Background() + + bucket := memblob.OpenBucket(nil) + defer bucket.Close() + + files := []string{ + "path/100%off/file.parquet", + "path/city=New York/file.parquet", + "path/to/file.parquet", + } + for _, f := range files { + require.NoError(t, bucket.WriteAll(ctx, f, []byte("data"), nil)) + } + + bfs := testADLSBlobFileIO(t, ctx, "abfs://container@account.dfs.core.windows.net/", bucket) + + var walked []string + err := bfs.WalkDir("abfs://container@account.dfs.core.windows.net/path", func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + + if !d.IsDir() { + walked = append(walked, path) + } + + return nil + }) + require.NoError(t, err) + + expected := []string{ + "abfs://container@account.dfs.core.windows.net/path/100%off/file.parquet", + "abfs://container@account.dfs.core.windows.net/path/city=New York/file.parquet", + "abfs://container@account.dfs.core.windows.net/path/to/file.parquet", + } + + assert.ElementsMatch(t, expected, walked) +} diff --git a/io/gocloud/azure/register.go b/io/gocloud/azure/register.go new file mode 100644 index 000000000..1bb0c6c64 --- /dev/null +++ b/io/gocloud/azure/register.go @@ -0,0 +1,47 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package azure provides the FileIO backend for Azure Data Lake Storage and Blob Storage. +// Import it for its side effects to register the abfs, abfss, wasb and wasbs schemes without linking the other clouds' +// SDKs: +// +// import _ "github.com/apache/iceberg-go/io/gocloud/azure" +package azure + +import ( + "context" + "net/url" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/io/gocloud/blobfs" +) + +func init() { + factory := func(ctx context.Context, parsed *url.URL, props map[string]string) (io.IO, error) { + bucket, err := createAzureBucket(ctx, parsed, props) + if err != nil { + return nil, err + } + + return blobfs.New(ctx, bucket, adlsObjectLocationExtractor(parsed)), nil + } + + for _, scheme := range schemes.Azure { + io.Register(scheme, factory) + } +} diff --git a/io/gocloud/azure/register_test.go b/io/gocloud/azure/register_test.go new file mode 100644 index 000000000..96fb5f4ce --- /dev/null +++ b/io/gocloud/azure/register_test.go @@ -0,0 +1,52 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package azure_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + _ "github.com/apache/iceberg-go/io/gocloud/azure" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRegistersOnlyItsOwnSchemes(t *testing.T) { + assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.Azure...), io.GetRegisteredSchemes()) +} + +func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { + ctx := context.Background() + + for _, tt := range []struct { + location string + errValue string + }{ + {"s3://bucket/key", "io/gocloud/s3"}, + {"s3a://bucket/key", "io/gocloud/s3"}, + {"gs://bucket/key", "io/gocloud/gcs"}, + } { + t.Run(tt.location, func(t *testing.T) { + _, err := io.LoadFS(ctx, nil, tt.location) + require.ErrorIs(t, err, io.ErrIOSchemeNotFound) + assert.ErrorContains(t, err, tt.errValue) + }) + } +} diff --git a/io/gocloud/blob.go b/io/gocloud/blobfs/blob.go similarity index 85% rename from io/gocloud/blob.go rename to io/gocloud/blobfs/blob.go index e4560e7d8..0a1010da0 100644 --- a/io/gocloud/blob.go +++ b/io/gocloud/blobfs/blob.go @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -package gocloud +// Package blobfs implements the iceberg-go/io FileIO interfaces on top of a gocloud.dev blob bucket. +// It links no cloud SDK of its own. +// The backend packages under io/gocloud each supply a bucket for their cloud. +package blobfs import ( "context" @@ -43,7 +46,7 @@ type blobOpenFile struct { *blob.Reader name, key string - b *BlobFileIO + b *FileIO ctx context.Context } @@ -83,7 +86,7 @@ var ErrEmptyObjectKey = errors.New("object key is empty") // URI to access it; this backend does not route across authorities. var ErrUnsupportedObjectAuthority = errors.New("object URI authority is not supported by this FileIO") -type objectLocation struct { +type ObjectLocation struct { scheme string authority string key string @@ -91,10 +94,10 @@ type objectLocation struct { hasAuthority bool } -func splitObjectLocation(location string) (objectLocation, error) { +func splitObjectLocation(location string) (ObjectLocation, error) { scheme, rest, ok := strings.Cut(location, "://") if !ok { - return objectLocation{key: location}, nil + return ObjectLocation{key: location}, nil } authorityEnd := strings.IndexAny(rest, "/?#") @@ -104,13 +107,13 @@ func splitObjectLocation(location string) (objectLocation, error) { authority := rest[:authorityEnd] if authority == "" { - return objectLocation{}, fmt.Errorf("URI authority is empty: %s", location) + return ObjectLocation{}, fmt.Errorf("URI authority is empty: %s", location) } key := "" if authorityEnd < len(rest) { if rest[authorityEnd] != '/' { - return objectLocation{}, fmt.Errorf("URI authority %q must be followed by an object path: %s", + return ObjectLocation{}, fmt.Errorf("URI authority %q must be followed by an object path: %s", authority, location) } @@ -119,18 +122,22 @@ func splitObjectLocation(location string) (objectLocation, error) { key = strings.TrimPrefix(rest[authorityEnd:], "/") } - return objectLocation{ + return NewObjectLocation(scheme, authority, key), nil +} + +func NewObjectLocation(scheme, authority, key string) ObjectLocation { + return ObjectLocation{ scheme: scheme, authority: authority, key: key, uriPrefix: scheme + "://" + authority + "/", hasAuthority: true, - }, nil + } } -type objectLocationExtractor func(location string) (objectLocation, error) +type ObjectLocationExtractor func(location string) (ObjectLocation, error) -func keyExtractorFromObjectLocation(extract objectLocationExtractor) KeyExtractor { +func KeyExtractorFromObjectLocation(extract ObjectLocationExtractor) KeyExtractor { return func(location string) (string, error) { parsed, err := extract(location) if err != nil { @@ -141,20 +148,20 @@ func keyExtractorFromObjectLocation(extract objectLocationExtractor) KeyExtracto } } -func defaultObjectLocationExtractor(bucketName string, allowedSchemes ...string) objectLocationExtractor { - return func(location string) (objectLocation, error) { +func DefaultObjectLocationExtractor(bucketName string, allowedSchemes ...string) ObjectLocationExtractor { + return func(location string) (ObjectLocation, error) { parsed, err := splitObjectLocation(location) if err != nil { - return objectLocation{}, err + return ObjectLocation{}, err } if parsed.hasAuthority { if len(allowedSchemes) > 0 && !slices.Contains(allowedSchemes, parsed.scheme) { - return objectLocation{}, fmt.Errorf("URI scheme %q is not supported by this FileIO (allowed: %s): %s", + return ObjectLocation{}, fmt.Errorf("URI scheme %q is not supported by this FileIO (allowed: %s): %s", parsed.scheme, strings.Join(allowedSchemes, ", "), location) } if parsed.authority != bucketName { - return objectLocation{}, fmt.Errorf("%w: URI authority %q does not match configured authority %q", + return ObjectLocation{}, fmt.Errorf("%w: URI authority %q does not match configured authority %q", ErrUnsupportedObjectAuthority, parsed.authority, bucketName) } @@ -173,13 +180,13 @@ func defaultObjectLocationExtractor(bucketName string, allowedSchemes ...string) // defaultKeyExtractor extracts the object key by removing the scheme and bucket name from the URI. // e.g., s3://bucket/path/file -> path/file. func defaultKeyExtractor(bucketName string, allowedSchemes ...string) KeyExtractor { - return keyExtractorFromObjectLocation(defaultObjectLocationExtractor(bucketName, allowedSchemes...)) + return KeyExtractorFromObjectLocation(DefaultObjectLocationExtractor(bucketName, allowedSchemes...)) } -type BlobFileIO struct { +type FileIO struct { *blob.Bucket - extractObject objectLocationExtractor + extractObject ObjectLocationExtractor ctx context.Context // newRangeReader is an optional hook for testing. @@ -187,7 +194,10 @@ type BlobFileIO struct { newRangeReader func(ctx context.Context, key string, offset, length int64) (io.ReadCloser, error) } -var _ icebergio.ListableIO = (*BlobFileIO)(nil) +var ( + _ icebergio.ListableIO = (*FileIO)(nil) + _ icebergio.BulkRemovableIO = (*FileIO)(nil) +) // deleteFilesMaxConcurrency bounds the number of in-flight object-store // deletes without creating one goroutine per path for large cleanup jobs. @@ -212,8 +222,8 @@ func (f blobFileInfo) IsDir() bool { return f.mode.IsDir() } func (f blobFileInfo) Sys() any { return f.sys } // preprocess returns the object key from an input path -func (bfs *BlobFileIO) preprocess(path string) (string, error) { - location, err := bfs.objectLocation(path) +func (bfs *FileIO) preprocess(path string) (string, error) { + location, err := bfs.resolveLocation(path) if err != nil { return "", err } @@ -253,7 +263,7 @@ func directoryName(key string) string { return pathpkg.Base(key) } -func (bfs *BlobFileIO) Open(path string) (icebergio.File, error) { +func (bfs *FileIO) Open(path string) (icebergio.File, error) { originalPath := path var err error path, err = bfs.preprocess(path) @@ -274,7 +284,7 @@ func (bfs *BlobFileIO) Open(path string) (icebergio.File, error) { return &blobOpenFile{Reader: r, name: name, key: key, b: bfs, ctx: bfs.ctx}, nil } -func (bfs *BlobFileIO) Remove(name string) error { +func (bfs *FileIO) Remove(name string) error { var err error name, err = bfs.preprocess(name) if err != nil { @@ -301,11 +311,11 @@ func (bfs *BlobFileIO) Remove(name string) error { return nil } -func (bfs *BlobFileIO) Create(name string) (icebergio.FileWriter, error) { +func (bfs *FileIO) Create(name string) (icebergio.FileWriter, error) { return bfs.NewWriter(bfs.ctx, name, true, nil) } -func (bfs *BlobFileIO) WriteFile(name string, content []byte) error { +func (bfs *FileIO) WriteFile(name string, content []byte) error { var err error name, err = bfs.preprocess(name) if err != nil { @@ -323,7 +333,7 @@ func (bfs *BlobFileIO) WriteFile(name string, content []byte) error { // // The caller must call Close on the returned Writer, even if the write is // aborted. -func (bfs *BlobFileIO) NewWriter(ctx context.Context, path string, overwrite bool, opts *blob.WriterOptions) (w *blobWriteFile, err error) { +func (bfs *FileIO) NewWriter(ctx context.Context, path string, overwrite bool, opts *blob.WriterOptions) (w *blobWriteFile, err error) { path, err = bfs.preprocess(path) if err != nil { return nil, &fs.PathError{Op: "new writer", Path: path, Err: err} @@ -353,19 +363,19 @@ func (bfs *BlobFileIO) NewWriter(ctx context.Context, path string, overwrite boo nil } -func createBlobFS(ctx context.Context, bucket *blob.Bucket, extractObject objectLocationExtractor) icebergio.IO { - return &BlobFileIO{Bucket: bucket, extractObject: extractObject, ctx: ctx} +func New(ctx context.Context, bucket *blob.Bucket, extractObject ObjectLocationExtractor) *FileIO { + return &FileIO{Bucket: bucket, extractObject: extractObject, ctx: ctx} } -func (bfs *BlobFileIO) objectLocation(root string) (objectLocation, error) { +func (bfs *FileIO) resolveLocation(root string) (ObjectLocation, error) { if bfs.extractObject == nil { - return objectLocation{}, errors.New("blob file IO missing object location extractor") + return ObjectLocation{}, errors.New("blob file IO missing object location extractor") } return bfs.extractObject(root) } -func walkedURIPath(location objectLocation, walked string) string { +func walkedURIPath(location ObjectLocation, walked string) string { if walked == "." { return location.uriPrefix } @@ -397,8 +407,8 @@ func isDirectoryMarker(walkRootKey string, dirEntry fs.DirEntry) bool { return ok && obj.Key == directoryMarker(walkRootKey) } -func (bfs *BlobFileIO) WalkDir(root string, fn fs.WalkDirFunc) error { - location, err := bfs.objectLocation(root) +func (bfs *FileIO) WalkDir(root string, fn fs.WalkDirFunc) error { + location, err := bfs.resolveLocation(root) var walkPath string if err != nil { if !errors.Is(err, ErrEmptyObjectKey) { @@ -425,7 +435,7 @@ func (bfs *BlobFileIO) WalkDir(root string, fn fs.WalkDirFunc) error { }) } -func (bfs *BlobFileIO) deleteFile(ctx context.Context, p string) (bool, error) { +func (bfs *FileIO) deleteFile(ctx context.Context, p string) (bool, error) { key, err := bfs.preprocess(p) if err != nil { return false, fmt.Errorf("failed to delete %s: %w", p, err) @@ -443,7 +453,7 @@ func (bfs *BlobFileIO) deleteFile(ctx context.Context, p string) (bool, error) { return true, nil } -func (bfs *BlobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]string, error) { +func (bfs *FileIO) DeleteFiles(ctx context.Context, paths []string) ([]string, error) { if len(paths) == 0 { return nil, nil } @@ -495,7 +505,7 @@ func (bfs *BlobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin } // MkdirAll mimics creating a directory by creating a zero-length object for each component of the path -func (bfs *BlobFileIO) MkdirAll(path string) error { +func (bfs *FileIO) MkdirAll(path string) error { key, err := bfs.preprocess(path) if err != nil { return &fs.PathError{Op: "mkdir", Path: path, Err: err} @@ -518,7 +528,7 @@ func (bfs *BlobFileIO) MkdirAll(path string) error { } // ReadFile reads the contents of the file at the given path and returns it as a byte slice. -func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) { +func (bfs *FileIO) ReadFile(path string) ([]byte, error) { key, err := bfs.preprocess(path) if err != nil { return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err} @@ -534,7 +544,7 @@ func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) { // Stat interprets the input path as a directory or file and returns the corresponding FileInfo. // If the path does not exist, it returns fs.ErrNotExist -func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) { +func (bfs *FileIO) Stat(path string) (fs.FileInfo, error) { key, err := bfs.preprocess(path) if err != nil { return nil, &fs.PathError{Op: "Stat", Path: path, Err: err} @@ -607,7 +617,7 @@ func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) { } // Rename renames one file from oldpath to newpath, replacing newpath if it already exists. -func (bfs *BlobFileIO) Rename(oldpath, newpath string) error { +func (bfs *FileIO) Rename(oldpath, newpath string) error { oldKey, err := bfs.preprocess(oldpath) if err != nil { return &fs.PathError{Op: "Rename", Path: oldpath, Err: err} @@ -630,7 +640,7 @@ func (bfs *BlobFileIO) Rename(oldpath, newpath string) error { } // RenameNoReplace renames one file/object (non-recursive) from oldpath to newpath, returning an error if newpath already exists. -func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error { +func (bfs *FileIO) RenameNoReplace(oldpath, newpath string) error { if _, err := bfs.Stat(newpath); err == nil { return &fs.PathError{Op: "RenameNoReplace", Path: newpath, Err: fs.ErrExist} // if the error is just that the file doesn't exist, we can continue with the rename @@ -644,7 +654,7 @@ func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error { // RemoveAll removes either a single file or interprets the path as and removes both // it and and all its children. -func (bfs *BlobFileIO) RemoveAll(name string) error { +func (bfs *FileIO) RemoveAll(name string) error { key, err := bfs.preprocess(name) if err != nil { return &fs.PathError{Op: "RemoveAll", Path: name, Err: err} @@ -681,7 +691,7 @@ func (bfs *BlobFileIO) RemoveAll(name string) error { type blobWriteFile struct { *blob.Writer name string - b *BlobFileIO + b *FileIO } func (f *blobWriteFile) Name() string { return f.name } diff --git a/io/gocloud/blob_test.go b/io/gocloud/blobfs/blob_test.go similarity index 85% rename from io/gocloud/blob_test.go rename to io/gocloud/blobfs/blob_test.go index 915d13f3d..609b8d08c 100644 --- a/io/gocloud/blob_test.go +++ b/io/gocloud/blobfs/blob_test.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package blobfs import ( "context" @@ -23,7 +23,6 @@ import ( "fmt" "io" "io/fs" - "net/url" "sync" "testing" "time" @@ -31,6 +30,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/apache/iceberg-go/internal/schemes" icebergio "github.com/apache/iceberg-go/io" "gocloud.dev/blob" "gocloud.dev/blob/driver" @@ -113,13 +113,13 @@ func TestDefaultKeyExtractor(t *testing.T) { }, { name: "s3 extractor rejects gs URI with same bucket", - allowedSchemes: s3Schemes, + allowedSchemes: schemes.S3, input: "gs://my-bucket/path/to/file.parquet", wantErrContains: `URI scheme "gs" is not supported`, }, { name: "gcs extractor rejects s3 URI with same bucket", - allowedSchemes: gcsSchemes, + allowedSchemes: schemes.GCS, input: "s3://my-bucket/path/to/file.parquet", wantErrContains: `URI scheme "s3" is not supported`, }, @@ -202,36 +202,21 @@ func TestBlobFileIOOpenPreprocessErrorRetainsOriginalPath(t *testing.T) { assert.Equal(t, name, pathErr.Path) } -func testBlobFileIO(ctx context.Context, bucketName string, bucket *blob.Bucket, allowedSchemes ...string) *BlobFileIO { +func testBlobFileIO(ctx context.Context, bucketName string, bucket *blob.Bucket, allowedSchemes ...string) *FileIO { if len(allowedSchemes) == 0 { - allowedSchemes = s3Schemes + allowedSchemes = schemes.S3 } - extractor := defaultObjectLocationExtractor(bucketName, allowedSchemes...) + extractor := DefaultObjectLocationExtractor(bucketName, allowedSchemes...) - return &BlobFileIO{ + return &FileIO{ Bucket: bucket, extractObject: extractor, ctx: ctx, } } -func testADLSBlobFileIO(t *testing.T, ctx context.Context, root string, bucket *blob.Bucket) *BlobFileIO { - t.Helper() - - parsed, err := url.Parse(root) - require.NoError(t, err) - - extractor := adlsObjectLocationExtractor(parsed) - - return &BlobFileIO{ - Bucket: bucket, - extractObject: extractor, - ctx: ctx, - } -} - -func identityObjectLocation(location string) (objectLocation, error) { - return objectLocation{key: location}, nil +func identityObjectLocation(location string) (ObjectLocation, error) { + return ObjectLocation{key: location}, nil } func TestBlobFileIORejectsUnsupportedObjectPaths(t *testing.T) { @@ -247,7 +232,7 @@ func TestBlobFileIORejectsUnsupportedObjectPaths(t *testing.T) { }{ { name: "s3 different bucket", - allowedSchemes: s3Schemes, + allowedSchemes: schemes.S3, path: "s3://other-bucket/data/file.parquet", oldKey: "other-bucket/data/file.parquet", wantErr: ErrUnsupportedObjectAuthority, @@ -255,7 +240,7 @@ func TestBlobFileIORejectsUnsupportedObjectPaths(t *testing.T) { }, { name: "gcs different bucket", - allowedSchemes: gcsSchemes, + allowedSchemes: schemes.GCS, path: "gs://other-bucket/data/file.parquet", oldKey: "other-bucket/data/file.parquet", wantErr: ErrUnsupportedObjectAuthority, @@ -263,14 +248,14 @@ func TestBlobFileIORejectsUnsupportedObjectPaths(t *testing.T) { }, { name: "s3 rejects gs same bucket", - allowedSchemes: s3Schemes, + allowedSchemes: schemes.S3, path: "gs://test-bucket/data/file.parquet", oldKey: "data/file.parquet", wantErrText: `URI scheme "gs" is not supported`, }, { name: "gcs rejects s3 same bucket", - allowedSchemes: gcsSchemes, + allowedSchemes: schemes.GCS, path: "s3://test-bucket/data/file.parquet", oldKey: "data/file.parquet", wantErrText: `URI scheme "s3" is not supported`, @@ -301,7 +286,7 @@ func TestNewWriterExistsError(t *testing.T) { bucket := memblob.OpenBucket(nil) - bfs := &BlobFileIO{ + bfs := &FileIO{ Bucket: bucket, extractObject: identityObjectLocation, ctx: ctx, @@ -369,7 +354,7 @@ func TestReadAtResourceCleanup(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var lastReaderClosed bool - bfs := &BlobFileIO{ + bfs := &FileIO{ Bucket: bucket, extractObject: identityObjectLocation, ctx: ctx, @@ -487,8 +472,8 @@ func TestBlobFileIOWalkDirRejectsWrongBucket(t *testing.T) { allowedSchemes []string root string }{ - {name: "s3", allowedSchemes: s3Schemes, root: "s3://other-bucket/"}, - {name: "gcs", allowedSchemes: gcsSchemes, root: "gs://other-bucket/data"}, + {name: "s3", allowedSchemes: schemes.S3, root: "s3://other-bucket/"}, + {name: "gcs", allowedSchemes: schemes.GCS, root: "gs://other-bucket/data"}, } { t.Run(tt.name, func(t *testing.T) { bfs := testBlobFileIO(ctx, "test-bucket", bucket, tt.allowedSchemes...) @@ -503,23 +488,6 @@ func TestBlobFileIOWalkDirRejectsWrongBucket(t *testing.T) { } } -func TestBlobFileIOWalkDirRejectsWrongAzureAuthority(t *testing.T) { - ctx := context.Background() - - bucket := memblob.OpenBucket(nil) - defer bucket.Close() - - bfs := testADLSBlobFileIO(t, ctx, "abfs://container@account.dfs.core.windows.net/", bucket) - - err := bfs.WalkDir("abfs://other@account.dfs.core.windows.net/data", func(string, fs.DirEntry, error) error { - t.Fatal("WalkDir callback should not be called") - - return nil - }) - require.ErrorContains(t, err, "does not match configured authority") - require.ErrorIs(t, err, ErrUnsupportedObjectAuthority) -} - func TestBlobFileIOWalkDirRelativeRootReturnsBareKeys(t *testing.T) { ctx := context.Background() @@ -583,45 +551,6 @@ func TestBlobFileIOWalkDirSubPath(t *testing.T) { assert.ElementsMatch(t, expected, walked) } -func TestBlobFileIOWalkDirAzureURI(t *testing.T) { - ctx := context.Background() - - bucket := memblob.OpenBucket(nil) - defer bucket.Close() - - files := []string{ - "path/100%off/file.parquet", - "path/city=New York/file.parquet", - "path/to/file.parquet", - } - for _, f := range files { - require.NoError(t, bucket.WriteAll(ctx, f, []byte("data"), nil)) - } - - bfs := testADLSBlobFileIO(t, ctx, "abfs://container@account.dfs.core.windows.net/", bucket) - - var walked []string - err := bfs.WalkDir("abfs://container@account.dfs.core.windows.net/path", func(path string, d fs.DirEntry, err error) error { - if err != nil { - return err - } - - if !d.IsDir() { - walked = append(walked, path) - } - - return nil - }) - require.NoError(t, err) - - expected := []string{ - "abfs://container@account.dfs.core.windows.net/path/100%off/file.parquet", - "abfs://container@account.dfs.core.windows.net/path/city=New York/file.parquet", - "abfs://container@account.dfs.core.windows.net/path/to/file.parquet", - } - assert.ElementsMatch(t, expected, walked) -} - func TestBlobFileIORawQueryFragmentRoundTrip(t *testing.T) { ctx := context.Background() @@ -642,17 +571,6 @@ func TestBlobFileIORawQueryFragmentRoundTrip(t *testing.T) { assert.Equal(t, content, got) } -func TestBlobFileIOImplementsBulkRemovableIO(t *testing.T) { - bucket := memblob.OpenBucket(nil) - defer bucket.Close() - - extractor := defaultObjectLocationExtractor("test-bucket") - bfs := createBlobFS(context.Background(), bucket, extractor) - - _, ok := bfs.(icebergio.BulkRemovableIO) - assert.True(t, ok, "blobFileIO should implement BulkRemovableIO") -} - func TestBlobFileIODeleteFiles(t *testing.T) { ctx := context.Background() bucket := memblob.OpenBucket(nil) @@ -663,9 +581,9 @@ func TestBlobFileIODeleteFiles(t *testing.T) { require.NoError(t, bucket.WriteAll(ctx, "data/file2.parquet", []byte("data2"), nil)) require.NoError(t, bucket.WriteAll(ctx, "data/file3.parquet", []byte("data3"), nil)) - extractor := defaultObjectLocationExtractor("test-bucket") - bfs := createBlobFS(ctx, bucket, extractor) - bulk := bfs.(icebergio.BulkRemovableIO) + extractor := DefaultObjectLocationExtractor("test-bucket") + bfs := New(ctx, bucket, extractor) + var bulk icebergio.BulkRemovableIO = bfs deleted, err := bulk.DeleteFiles(ctx, []string{ "s3://test-bucket/data/file1.parquet", @@ -693,9 +611,9 @@ func TestBlobFileIODeleteFilesMissingFilesAreNotErrors(t *testing.T) { bucket := memblob.OpenBucket(nil) defer bucket.Close() - extractor := defaultObjectLocationExtractor("test-bucket") - bfs := createBlobFS(ctx, bucket, extractor) - bulk := bfs.(icebergio.BulkRemovableIO) + extractor := DefaultObjectLocationExtractor("test-bucket") + bfs := New(ctx, bucket, extractor) + var bulk icebergio.BulkRemovableIO = bfs // Deleting non-existent files should succeed. deleted, err := bulk.DeleteFiles(ctx, []string{ @@ -710,8 +628,8 @@ func TestBlobFileIORemoveMissingFileReturnsNotExist(t *testing.T) { bucket := memblob.OpenBucket(nil) defer bucket.Close() - extractor := defaultObjectLocationExtractor("test-bucket") - bfs := createBlobFS(ctx, bucket, extractor) + extractor := DefaultObjectLocationExtractor("test-bucket") + bfs := New(ctx, bucket, extractor) err := bfs.Remove("s3://test-bucket/data/nonexistent.parquet") require.ErrorIs(t, err, fs.ErrNotExist) @@ -722,9 +640,9 @@ func TestBlobFileIODeleteFilesEmpty(t *testing.T) { bucket := memblob.OpenBucket(nil) defer bucket.Close() - extractor := defaultObjectLocationExtractor("test-bucket") - bfs := createBlobFS(ctx, bucket, extractor) - bulk := bfs.(icebergio.BulkRemovableIO) + extractor := DefaultObjectLocationExtractor("test-bucket") + bfs := New(ctx, bucket, extractor) + var bulk icebergio.BulkRemovableIO = bfs deleted, err := bulk.DeleteFiles(ctx, nil) require.NoError(t, err) @@ -808,9 +726,9 @@ func TestBlobFileIODeleteFilesIsConcurrentAndBounded(t *testing.T) { bucket := blob.NewBucket(tracker) defer bucket.Close() - bfs := &BlobFileIO{ + bfs := &FileIO{ Bucket: bucket, - extractObject: defaultObjectLocationExtractor("test-bucket"), + extractObject: DefaultObjectLocationExtractor("test-bucket"), ctx: context.Background(), } paths := make([]string, pathCount) @@ -863,7 +781,7 @@ func TestBlobFileIOStat(t *testing.T) { require.NoError(t, bucket.WriteAll(ctx, "data/file.parquet", []byte("content"), nil)) - bfs := createBlobFS(ctx, bucket, defaultObjectLocationExtractor("test-bucket")).(*BlobFileIO) + bfs := New(ctx, bucket, DefaultObjectLocationExtractor("test-bucket")) fileInfo, err := bfs.Stat("s3://test-bucket/data/file.parquet") require.NoError(t, err) @@ -897,7 +815,7 @@ func TestBlobFileIOMkdirAll(t *testing.T) { bucket := memblob.OpenBucket(nil) defer bucket.Close() - bfs := createBlobFS(ctx, bucket, defaultObjectLocationExtractor("test-bucket")).(*BlobFileIO) + bfs := New(ctx, bucket, DefaultObjectLocationExtractor("test-bucket")) require.NoError(t, bfs.MkdirAll("s3://test-bucket/a/b/c")) @@ -913,7 +831,7 @@ func TestBlobFileIOWalkDirSkipsDirectoryMarker(t *testing.T) { bucket := memblob.OpenBucket(nil) defer bucket.Close() - bfs := createBlobFS(ctx, bucket, defaultObjectLocationExtractor("test-bucket")).(*BlobFileIO) + bfs := New(ctx, bucket, DefaultObjectLocationExtractor("test-bucket")) // MkdirAll leaves only a "warehouse/ns/" marker for an empty namespace. // Walking "warehouse/ns" should not report that marker as a child file. @@ -941,7 +859,7 @@ func TestBlobFileIORemoveAll(t *testing.T) { require.NoError(t, bucket.WriteAll(ctx, "warehouse/ns/tbl/data/00001.parquet", []byte("data"), nil)) require.NoError(t, bucket.WriteAll(ctx, "warehouse/other/keep.parquet", []byte("keep"), nil)) - bfs := createBlobFS(ctx, bucket, defaultObjectLocationExtractor("test-bucket")).(*BlobFileIO) + bfs := New(ctx, bucket, DefaultObjectLocationExtractor("test-bucket")) require.NoError(t, bfs.RemoveAll("s3://test-bucket/data/file.parquet")) exists, err := bucket.Exists(ctx, "data/file.parquet") diff --git a/io/gocloud/utils.go b/io/gocloud/blobfs/utils.go similarity index 92% rename from io/gocloud/utils.go rename to io/gocloud/blobfs/utils.go index 7067e3858..9c5315bc8 100644 --- a/io/gocloud/utils.go +++ b/io/gocloud/blobfs/utils.go @@ -15,11 +15,11 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package blobfs import "strings" -func propertiesWithPrefix(props map[string]string, prefix string) map[string]string { +func PropertiesWithPrefix(props map[string]string, prefix string) map[string]string { result := map[string]string{} for k, v := range props { if after, ok := strings.CutPrefix(k, prefix); ok { diff --git a/io/gocloud/compat_test.go b/io/gocloud/compat_test.go new file mode 100644 index 000000000..4bd3a8219 --- /dev/null +++ b/io/gocloud/compat_test.go @@ -0,0 +1,100 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package gocloud_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/io/gocloud" + "github.com/apache/iceberg-go/io/gocloud/blobfs" + "github.com/apache/iceberg-go/io/gocloud/gcs" + "github.com/apache/iceberg-go/io/gocloud/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// For callers who still have not migrated +var ( + _ *blobfs.FileIO = (*gocloud.BlobFileIO)(nil) + _ blobfs.KeyExtractor = gocloud.KeyExtractor(nil) + _ error = gocloud.ErrEmptyObjectKey + _ error = gocloud.ErrUnsupportedObjectAuthority +) + +func TestDeprecatedParseConfigWrappers(t *testing.T) { + ctx := context.Background() + + t.Run("ParseGCSConfig", func(t *testing.T) { + for _, props := range []map[string]string{ + {}, + {io.GCSUseJSONAPI: "true"}, + {io.GCSUseJSONAPI: "false"}, + {io.GCSUseJSONAPI: "not-a-bool"}, + {io.GCSEndpoint: "http://localhost:4443"}, + {io.GCSEndpoint: "http://localhost:4443", io.GCSUseJSONAPI: "true"}, + } { + want, got := gcs.ParseGCSConfig(props), gocloud.ParseGCSConfig(props) + assert.Equal(t, want, got, "props: %v", props) + } + }) + + t.Run("ParseAWSConfig", func(t *testing.T) { + for _, tt := range []struct { + props map[string]string + static bool + }{ + {props: map[string]string{}}, + {props: map[string]string{io.S3Region: "us-west-2"}}, + {props: map[string]string{io.S3ClientRegion: "eu-central-1"}}, + {props: map[string]string{io.S3Region: "us-west-2", io.S3ClientRegion: "eu-central-1"}}, + {props: map[string]string{"token": "bearer-token"}}, + {props: map[string]string{io.S3AccessKeyID: "ak", io.S3SecretAccessKey: "sk"}, static: true}, + {props: map[string]string{io.S3AccessKeyID: "ak", io.S3SecretAccessKey: "sk", io.S3SessionToken: "st"}, static: true}, + } { + want, wantErr := s3.ParseAWSConfig(ctx, tt.props) + got, gotErr := gocloud.ParseAWSConfig(ctx, tt.props) + + require.NoError(t, wantErr, "props: %v", tt.props) + require.NoError(t, gotErr, "props: %v", tt.props) + assert.Equal(t, want.Region, got.Region, "props: %v", tt.props) + + // Retrieving from the default chain would reach the network, so + // only static credentials are compared by value. + if !tt.static { + continue + } + + wantCreds, err := want.Credentials.Retrieve(ctx) + require.NoError(t, err) + gotCreds, err := got.Credentials.Retrieve(ctx) + require.NoError(t, err) + assert.Equal(t, wantCreds, gotCreds, "props: %v", tt.props) + } + }) + + t.Run("error is relayed", func(t *testing.T) { + props := map[string]string{io.S3RemoteSigningEnabled: "true"} + _, wantErr := s3.ParseAWSConfig(ctx, props) + _, gotErr := gocloud.ParseAWSConfig(ctx, props) + require.Error(t, wantErr) + require.Error(t, gotErr) + assert.Equal(t, wantErr.Error(), gotErr.Error()) + }) +} diff --git a/io/gocloud/gcs.go b/io/gocloud/gcs/gcs.go similarity index 99% rename from io/gocloud/gcs.go rename to io/gocloud/gcs/gcs.go index 1b01550ba..25f9af4d3 100644 --- a/io/gocloud/gcs.go +++ b/io/gocloud/gcs/gcs.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package gcs import ( "context" diff --git a/io/gocloud/gcs_integration_test.go b/io/gocloud/gcs/gcs_integration_test.go similarity index 98% rename from io/gocloud/gcs_integration_test.go rename to io/gocloud/gcs/gcs_integration_test.go index ebaba81fc..b7630fa35 100644 --- a/io/gocloud/gcs_integration_test.go +++ b/io/gocloud/gcs/gcs_integration_test.go @@ -17,7 +17,7 @@ //go:build integration -package gocloud_test +package gcs_test import ( "bytes" @@ -32,7 +32,7 @@ import ( "github.com/apache/iceberg-go/catalog" sqlcat "github.com/apache/iceberg-go/catalog/sql" "github.com/apache/iceberg-go/io" - _ "github.com/apache/iceberg-go/io/gocloud" + _ "github.com/apache/iceberg-go/io/gocloud/gcs" "github.com/stretchr/testify/suite" "github.com/uptrace/bun/driver/sqliteshim" ) diff --git a/io/gocloud/gcs_test.go b/io/gocloud/gcs/gcs_test.go similarity index 99% rename from io/gocloud/gcs_test.go rename to io/gocloud/gcs/gcs_test.go index 6b3e7a66e..939d9c35b 100644 --- a/io/gocloud/gcs_test.go +++ b/io/gocloud/gcs/gcs_test.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package gcs import ( "context" diff --git a/io/gocloud/gcs/register.go b/io/gocloud/gcs/register.go new file mode 100644 index 000000000..f72d91d4f --- /dev/null +++ b/io/gocloud/gcs/register.go @@ -0,0 +1,47 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package gcs provides the FileIO backend for Google Cloud Storage. +// Import it for its side effects to register the gs schemes without linking the other clouds' +// SDKs: +// +// import _ "github.com/apache/iceberg-go/io/gocloud/gcs" +package gcs + +import ( + "context" + "net/url" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/io/gocloud/blobfs" +) + +func init() { + factory := func(ctx context.Context, parsed *url.URL, props map[string]string) (io.IO, error) { + bucket, err := createGCSBucket(ctx, parsed, props) + if err != nil { + return nil, err + } + + return blobfs.New(ctx, bucket, blobfs.DefaultObjectLocationExtractor(parsed.Host, schemes.GCS...)), nil + } + + for _, scheme := range schemes.GCS { + io.Register(scheme, factory) + } +} diff --git a/io/gocloud/gcs/register_test.go b/io/gocloud/gcs/register_test.go new file mode 100644 index 000000000..7897f082e --- /dev/null +++ b/io/gocloud/gcs/register_test.go @@ -0,0 +1,52 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package gcs_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + _ "github.com/apache/iceberg-go/io/gocloud/gcs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRegistersOnlyItsOwnSchemes(t *testing.T) { + assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.GCS...), io.GetRegisteredSchemes()) +} + +func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { + ctx := context.Background() + + for _, tt := range []struct { + location string + wantHint string + }{ + {"s3://bucket/key", "io/gocloud/s3"}, + {"oss://bucket/key", "io/gocloud/s3"}, + {"abfs://container@account.dfs.core.windows.net/key", "io/gocloud/azure"}, + } { + t.Run(tt.location, func(t *testing.T) { + _, err := io.LoadFS(ctx, nil, tt.location) + require.ErrorIs(t, err, io.ErrIOSchemeNotFound) + assert.ErrorContains(t, err, tt.wantHint) + }) + } +} diff --git a/io/gocloud/gocloud.go b/io/gocloud/gocloud.go new file mode 100644 index 000000000..a53a1cbb2 --- /dev/null +++ b/io/gocloud/gocloud.go @@ -0,0 +1,66 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package gocloud registers every gocloud.dev-backed FileIO implementation and +// therefore links the AWS, Google Cloud and Azure SDKs. +// To link only the clouds an application uses, blank-import io/gocloud/s3, io/gocloud/gcs or +// io/gocloud/azure instead, in any combination. +package gocloud + +import ( + "context" + + "github.com/apache/iceberg-go/io/gocloud/blobfs" + "github.com/aws/aws-sdk-go-v2/aws" + "gocloud.dev/blob/gcsblob" + + _ "github.com/apache/iceberg-go/io/gocloud/azure" + "github.com/apache/iceberg-go/io/gocloud/gcs" + "github.com/apache/iceberg-go/io/gocloud/s3" +) + +var ( + // Deprecated: use [blobfs.ErrEmptyObjectKey] + ErrEmptyObjectKey = blobfs.ErrEmptyObjectKey + // Deprecated: use [blobfs.ErrUnsupportedObjectAuthority] + ErrUnsupportedObjectAuthority = blobfs.ErrUnsupportedObjectAuthority +) + +type ( + // BlobFileIO is the FileIO implementation backed by a gocloud.dev bucket. + // + // Deprecated: use [blobfs.FileIO] + BlobFileIO = blobfs.FileIO + // KeyExtractor extracts the object key from an input path. + // + // Deprecated: use [blobfs.KeyExtractor] + KeyExtractor = blobfs.KeyExtractor +) + +// ParseAWSConfig parses the S3 properties and returns a configuration. +// +// Deprecated: use [s3.ParseAWSConfig] +func ParseAWSConfig(ctx context.Context, props map[string]string) (*aws.Config, error) { + return s3.ParseAWSConfig(ctx, props) +} + +// ParseGCSConfig parses GCS properties and returns bucket options. +// +// Deprecated: use [gcs.ParseGCSConfig] +func ParseGCSConfig(props map[string]string) *gcsblob.Options { + return gcs.ParseGCSConfig(props) +} diff --git a/io/gocloud/isolation_test.go b/io/gocloud/isolation_test.go new file mode 100644 index 000000000..d1eb7543d --- /dev/null +++ b/io/gocloud/isolation_test.go @@ -0,0 +1,99 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package gocloud_test + +import ( + "os/exec" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var cloudSDKPrefixes = map[string][]string{ + "aws": {"github.com/aws/aws-sdk-go-v2/"}, + "gcp": {"cloud.google.com/go/storage", "gocloud.dev/blob/gcsblob"}, + "azure": {"github.com/Azure/"}, +} + +// A package that pulls in a cloud's SDK still registers only its own schemes, +// thus, the registry cannot catch it. Inspect the build graph instead. +// +// catalog/hadoop is listed because it needs only the bucket-backed FileIO type; +// importing io/gocloud for that assertion would link all three SDKs into every +// binary that uses the Hadoop catalog. +func TestPackagesLinkOnlyTheirOwnCloudSDK(t *testing.T) { + for _, tt := range []struct { + pkg string + cloud string + }{ + {"./io/gocloud/blobfs", ""}, + {"./io/gocloud/s3", "aws"}, + {"./io/gocloud/gcs", "gcp"}, + {"./io/gocloud/azure", "azure"}, + {"./catalog/hadoop", ""}, + } { + t.Run(tt.pkg, func(t *testing.T) { + deps := packageDeps(t, tt.pkg) + + for cloud, prefixes := range cloudSDKPrefixes { + linked := depsWithAnyPrefix(deps, prefixes) + if cloud == tt.cloud { + assert.NotEmpty(t, linked, "%s should link the %s SDK", tt.pkg, cloud) + + continue + } + + assert.Empty(t, linked, "%s must not link the %s SDK", tt.pkg, cloud) + } + }) + } +} + +func packageDeps(t *testing.T, pkg string) []string { + t.Helper() + + goBin, err := exec.LookPath("go") + if err != nil { + t.Skipf("go toolchain not available: %v", err) + } + + cmd := exec.Command(goBin, "list", "-deps", pkg) + cmd.Dir = "../.." + + out, err := cmd.Output() + require.NoError(t, err, "go list -deps %s", pkg) + + return strings.Fields(string(out)) +} + +func depsWithAnyPrefix(deps, prefixes []string) []string { + var matched []string + for _, dep := range deps { + for _, prefix := range prefixes { + if strings.HasPrefix(dep, prefix) { + matched = append(matched, dep) + + break + } + } + } + + return matched +} diff --git a/io/gocloud/register.go b/io/gocloud/register.go deleted file mode 100644 index 98f1b3dd0..000000000 --- a/io/gocloud/register.go +++ /dev/null @@ -1,86 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -package gocloud - -import ( - "context" - "net/url" - - icebergio "github.com/apache/iceberg-go/io" -) - -func init() { - registerS3Schemes() - registerGCSScheme() - registerAzureSchemes() -} - -var ( - s3Schemes = []string{"s3", "s3a", "s3n", "oss"} - gcsSchemes = []string{"gs"} -) - -// registerS3Schemes registers S3-compatible storage schemes (s3, s3a, s3n). -func registerS3Schemes() { - s3Factory := func(ctx context.Context, parsed *url.URL, props map[string]string) (icebergio.IO, error) { - bucket, err := createS3Bucket(ctx, parsed, props) - if err != nil { - return nil, err - } - - extractor := defaultObjectLocationExtractor(parsed.Host, s3Schemes...) - - return createBlobFS(ctx, bucket, extractor), nil - } - icebergio.Register("s3", s3Factory) - icebergio.Register("s3a", s3Factory) - icebergio.Register("s3n", s3Factory) - icebergio.Register("oss", s3Factory) -} - -// registerGCSScheme registers the Google Cloud Storage scheme (gs). -func registerGCSScheme() { - icebergio.Register("gs", func(ctx context.Context, parsed *url.URL, props map[string]string) (icebergio.IO, error) { - bucket, err := createGCSBucket(ctx, parsed, props) - if err != nil { - return nil, err - } - - extractor := defaultObjectLocationExtractor(parsed.Host, gcsSchemes...) - - return createBlobFS(ctx, bucket, extractor), nil - }) -} - -// registerAzureSchemes registers Azure Data Lake Storage schemes (abfs, abfss, wasb, wasbs). -func registerAzureSchemes() { - azureFactory := func(ctx context.Context, parsed *url.URL, props map[string]string) (icebergio.IO, error) { - bucket, err := createAzureBucket(ctx, parsed, props) - if err != nil { - return nil, err - } - - extractor := adlsObjectLocationExtractor(parsed) - - return createBlobFS(ctx, bucket, extractor), nil - } - icebergio.Register("abfs", azureFactory) - icebergio.Register("abfss", azureFactory) - icebergio.Register("wasb", azureFactory) - icebergio.Register("wasbs", azureFactory) -} diff --git a/io/gocloud/s3/register.go b/io/gocloud/s3/register.go new file mode 100644 index 000000000..8e59b8dc4 --- /dev/null +++ b/io/gocloud/s3/register.go @@ -0,0 +1,47 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package s3 provides the FileIO backend for S3 and S3-compatible object stores. +// Import it for its side effects to register the s3, s3a, s3n and oss schemes without linking the other clouds' +// SDKs: +// +// import _ "github.com/apache/iceberg-go/io/gocloud/s3" +package s3 + +import ( + "context" + "net/url" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/io/gocloud/blobfs" +) + +func init() { + factory := func(ctx context.Context, parsed *url.URL, props map[string]string) (io.IO, error) { + bucket, err := createS3Bucket(ctx, parsed, props) + if err != nil { + return nil, err + } + + return blobfs.New(ctx, bucket, blobfs.DefaultObjectLocationExtractor(parsed.Host, schemes.S3...)), nil + } + + for _, scheme := range schemes.S3 { + io.Register(scheme, factory) + } +} diff --git a/io/gocloud/s3/register_test.go b/io/gocloud/s3/register_test.go new file mode 100644 index 000000000..54c6ca384 --- /dev/null +++ b/io/gocloud/s3/register_test.go @@ -0,0 +1,52 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package s3_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go/internal/schemes" + "github.com/apache/iceberg-go/io" + _ "github.com/apache/iceberg-go/io/gocloud/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRegistersOnlyItsOwnSchemes(t *testing.T) { + assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.S3...), io.GetRegisteredSchemes()) +} + +func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { + ctx := context.Background() + + for _, tt := range []struct { + location string + wantHint string + }{ + {"gs://bucket/key", "io/gocloud/gcs"}, + {"abfs://container@account.dfs.core.windows.net/key", "io/gocloud/azure"}, + {"wasbs://container@account.blob.core.windows.net/key", "io/gocloud/azure"}, + } { + t.Run(tt.location, func(t *testing.T) { + _, err := io.LoadFS(ctx, nil, tt.location) + require.ErrorIs(t, err, io.ErrIOSchemeNotFound) + assert.ErrorContains(t, err, tt.wantHint) + }) + } +} diff --git a/io/gocloud/s3.go b/io/gocloud/s3/s3.go similarity index 99% rename from io/gocloud/s3.go rename to io/gocloud/s3/s3.go index 55d792ecc..eb96ca0f3 100644 --- a/io/gocloud/s3.go +++ b/io/gocloud/s3/s3.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package s3 import ( "context" diff --git a/io/gocloud/s3_integration_test.go b/io/gocloud/s3/s3_integration_test.go similarity index 98% rename from io/gocloud/s3_integration_test.go rename to io/gocloud/s3/s3_integration_test.go index c3197edda..3bae04c11 100644 --- a/io/gocloud/s3_integration_test.go +++ b/io/gocloud/s3/s3_integration_test.go @@ -17,7 +17,7 @@ //go:build integration -package gocloud_test +package s3_test import ( "context" @@ -29,7 +29,7 @@ import ( "github.com/apache/iceberg-go/catalog" sqlcat "github.com/apache/iceberg-go/catalog/sql" "github.com/apache/iceberg-go/io" - _ "github.com/apache/iceberg-go/io/gocloud" + _ "github.com/apache/iceberg-go/io/gocloud/s3" "github.com/stretchr/testify/require" "github.com/uptrace/bun/driver/sqliteshim" ) diff --git a/io/gocloud/s3_test.go b/io/gocloud/s3/s3_test.go similarity index 99% rename from io/gocloud/s3_test.go rename to io/gocloud/s3/s3_test.go index 6384804d6..efd2fbd60 100644 --- a/io/gocloud/s3_test.go +++ b/io/gocloud/s3/s3_test.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package gocloud +package s3 import ( "context" diff --git a/io/io.go b/io/io.go index eecaa928c..35d3be3cb 100644 --- a/io/io.go +++ b/io/io.go @@ -24,9 +24,9 @@ // // import _ "github.com/apache/iceberg-go/io/gocloud" // -// Will register cloud storage implementations for S3, GCS, Azure, and in-memory -// blob storage. The local filesystem (file:// and empty scheme) is registered -// by default. +// Will register S3, GCS and Azure, and link all three cloud SDKs. +// Importing io/gocloud/s3, io/gocloud/gcs or io/gocloud/azure instead registers only that backend. +// The local filesystem (file:// and empty scheme) is always registered. package io import ( @@ -341,9 +341,9 @@ func (f ioFile) ReadDir(count int) ([]fs.DirEntry, error) { // (file:// or empty scheme) is registered by default. // // Additional schemes can be registered by importing subpackages. -// For S3, GCS, Azure and in-memory support, import: +// Import the backend for each cloud in use, or io/gocloud for all of them: // -// import _ "github.com/apache/iceberg-go/io/gocloud" +// import _ "github.com/apache/iceberg-go/io/gocloud/s3" func LoadFS(ctx context.Context, props map[string]string, location string) (IO, error) { if location == "" { location = props["warehouse"] diff --git a/io/registry.go b/io/registry.go index aacea7c4a..465352e43 100644 --- a/io/registry.go +++ b/io/registry.go @@ -24,6 +24,8 @@ import ( "net/url" "slices" "sync" + + "github.com/apache/iceberg-go/internal/schemes" ) type registry map[string]SchemeFactory @@ -75,13 +77,18 @@ func init() { Register("", localFSFactory) } +func schemeImportHint(backend string) string { + return "hint: import the matching IO module for side-effect registration: " + + `_ "github.com/apache/iceberg-go/io/gocloud/` + backend + `"` +} + func schemeRegistrationHint(scheme string) string { - switch scheme { - case "s3", "s3a", "s3n", "gs", "abfs", "abfss", "wasb", "wasbs": - return `hint: import the matching IO module for side-effect registration: _ "github.com/apache/iceberg-go/io/gocloud" // for s3/gcs/azblob` - default: + backend := schemes.BackendFor(scheme) + if backend == "" { return "" } + + return schemeImportHint(backend) } func inferFileIOFromScheme(ctx context.Context, path string, props map[string]string) (IO, error) { diff --git a/website/src/configuration.md b/website/src/configuration.md index 1b128a02c..f83b48e7e 100644 --- a/website/src/configuration.md +++ b/website/src/configuration.md @@ -151,16 +151,20 @@ Operations that create or update tables/views accept these (`catalog/catalog.go` iceberg-go registers the local file system (`file://`) automatically. Cloud schemes are *not* registered until you add a blank import: ```go -import _ "github.com/apache/iceberg-go/io/gocloud" +import _ "github.com/apache/iceberg-go/io/gocloud/s3" // s3, s3a, s3n, oss +import _ "github.com/apache/iceberg-go/io/gocloud/gcs" // gs +import _ "github.com/apache/iceberg-go/io/gocloud/azure" // abfs, abfss, wasb, wasbs ``` -The `init()` function in [`io/gocloud/register.go`](https://github.com/apache/iceberg-go/blob/main/io/gocloud/register.go) registers `s3`, `s3a`, `s3n`, `oss`, `gs`, `abfs`, `abfss`, `wasb`, and `wasbs`. Without the blank import, these schemes return `ErrIOSchemeNotFound` with a hint to add the import. +Each backend package links its own cloud SDK. Importing `io/gocloud` registers all three at once and links all cloud SDKs. + +Without a matching blank import, these schemes return `ErrIOSchemeNotFound` with a hint naming the package to import. All credential and tuning property keys are constants in [`io/config.go`](https://github.com/apache/iceberg-go/blob/main/io/config.go). They can be supplied through table properties, catalog properties, or per-call `iceberg.Properties` arguments depending on context. ### S3 -Authentication is resolved in this order (`io/gocloud/s3.go`): +Authentication is resolved in this order (`io/gocloud/s3/s3.go`): 1. Static credentials in properties: `s3.access-key-id` + `s3.secret-access-key` (+ optional `s3.session-token`). 2. The standard AWS SDK v2 default credential chain - environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`), `~/.aws/credentials`, container/IAM role. @@ -182,7 +186,7 @@ Tuning properties: ### Google Cloud Storage -Authentication resolution (`io/gocloud/gcs.go`): +Authentication resolution (`io/gocloud/gcs/gcs.go`): 1. Explicit JSON key bytes via `gcs.jsonkey` or path via `gcs.keypath`. 2. Optional `gcs.credtype` selecting one of `service_account`, `authorized_user`, `impersonated_service_account`, `external_account`. @@ -200,7 +204,7 @@ Tuning properties: ### Azure Data Lake Storage / Blob -Authentication is selected based on the property keys present (`io/gocloud/azure.go`): +Authentication is selected based on the property keys present (`io/gocloud/azure/azure.go`): 1. Shared key: both `adls.auth.shared-key.account.name` and `adls.auth.shared-key.account.key` set. 2. Per-host SAS token: `adls.sas-token.` (prefix-matched against the storage account host). @@ -228,7 +232,7 @@ iceberg-go reads only a small set of environment variables directly. AWS / GCP / |---|---|---| | `GOICEBERG_HOME` | Directory containing `.iceberg-go.yaml`. Defaults to the user's home directory. | `config/config.go:87` | | `ICEBERG_SQL_DEBUG` | SQL catalog query logging - `1` (failed queries), `2` (all queries). | `catalog/sql/sql.go:206` | -| `AWS_S3_ENDPOINT` | Fallback S3 endpoint when `s3.endpoint` is unset. | `io/gocloud/s3.go:193` | +| `AWS_S3_ENDPOINT` | Fallback S3 endpoint when `s3.endpoint` is unset. | `io/gocloud/s3/s3.go` | There is no `PYICEBERG_*`-style env var convention. Use the YAML config file or pass `iceberg.Properties` to overrides programmatically. @@ -264,7 +268,8 @@ func init() { } ``` -`io.Register` panics on `nil` factory or duplicate scheme. Built-in schemes: `file`, `""` (the empty scheme). Cloud schemes (`s3`, `gs`, `abfs`, etc.) are registered by `io/gocloud` only when its package is blank-imported. +`io.Register` panics on `nil` factory or duplicate scheme. Built-in schemes: `file`, `""` (the empty scheme). +Cloud schemes (`s3`, `gs`, `abfs`, etc.) are registered only when the matching backend package under `io/gocloud` is blank-imported. `io.GetRegisteredSchemes()` returns the current scheme list; `io.Unregister(scheme)` removes one. diff --git a/website/src/feature-status.md b/website/src/feature-status.md index feaafafbb..87d81986f 100644 --- a/website/src/feature-status.md +++ b/website/src/feature-status.md @@ -67,7 +67,7 @@ All V1 features are supported. V1 is the format-version baseline. | Azure Blob Storage | X | | Local Filesystem | X | -S3, GCS, and Azure require a blank import: `_ "github.com/apache/iceberg-go/io/gocloud"`. See [Configuration](./configuration.md). +S3, GCS, and Azure each require a blank import of their backend package, for example `_ "github.com/apache/iceberg-go/io/gocloud/s3"`. See [Configuration](./configuration.md). ## Metadata operations diff --git a/website/src/getting-started.md b/website/src/getting-started.md index e2ed1db5c..fc871da84 100644 --- a/website/src/getting-started.md +++ b/website/src/getting-started.md @@ -38,7 +38,7 @@ go get github.com/apache/arrow-go/v18@latest go get github.com/uptrace/bun/driver/sqliteshim@latest ``` -`iceberg-go` itself only registers the local file system. For S3, GCS, or Azure Blob you would also blank-import `github.com/apache/iceberg-go/io/gocloud`. We are staying on local disk for this tutorial. +`iceberg-go` itself only registers the local file system. For S3, GCS, or Azure Blob you would also blank-import the matching backend package, such as `github.com/apache/iceberg-go/io/gocloud/s3`. We are staying on local disk for this tutorial. ## 2. Open a local catalog From 4b80d2d6113443c2477b26b975bc6037f444d2fa Mon Sep 17 00:00:00 2001 From: badalprasadsingh Date: Mon, 31 Aug 2026 08:43:51 +0530 Subject: [PATCH 2/3] fix: minor Signed-off-by: badalprasadsingh --- catalog/hadoop/hadoop.go | 6 +++ catalog/hadoop/io_test.go | 58 ++++++++++++++++++++++++++ internal/schemes/schemes.go | 26 ++++++------ io/gocloud/azure/azure.go | 4 +- io/gocloud/azure/register_test.go | 12 ++++-- io/gocloud/{blobfs => azure}/utils.go | 4 +- io/gocloud/blobfs/blob.go | 60 +++++++++++++++++++-------- io/gocloud/blobfs/blob_test.go | 29 +++++++++++++ io/gocloud/compat_test.go | 8 ++++ io/gocloud/gcs/register_test.go | 8 +++- io/gocloud/s3/register_test.go | 8 +++- 11 files changed, 184 insertions(+), 39 deletions(-) create mode 100644 catalog/hadoop/io_test.go rename io/gocloud/{blobfs => azure}/utils.go (92%) diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go index 98dedbe35..76b93a55a 100644 --- a/catalog/hadoop/hadoop.go +++ b/catalog/hadoop/hadoop.go @@ -15,6 +15,12 @@ // specific language governing permissions and limitations // under the License. +// Package hadoop implements a catalog over a Hadoop-style warehouse directory. +// +// It links no cloud SDK. Callers on cloud storage must blank-import the backend for their scheme. +// For example io/gocloud/s3. +// +// Otherwise the first LoadFS for a cloud path fails with io.ErrIOSchemeNotFound naming the package to import. package hadoop import ( diff --git a/catalog/hadoop/io_test.go b/catalog/hadoop/io_test.go new file mode 100644 index 000000000..86a6ce1cb --- /dev/null +++ b/catalog/hadoop/io_test.go @@ -0,0 +1,58 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// The integration build blank-imports io/gocloud, which registers the schemes +// this file asserts are absent. +//go:build !integration + +package hadoop + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go/internal/schemes" + icebergio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestImportRegistersNoCloudSchemes(t *testing.T) { + registered := icebergio.GetRegisteredSchemes() + for _, list := range [][]string{schemes.S3, schemes.GCS, schemes.Azure} { + for _, scheme := range list { + assert.NotContains(t, registered, scheme) + } + } +} + +func TestLoadFSCloudPathReportsMissingBackend(t *testing.T) { + for _, tt := range []struct { + location string + wantHint string + }{ + {"s3://bucket/key", "io/gocloud/s3"}, + {"gs://bucket/key", "io/gocloud/gcs"}, + {"abfs://container@account.dfs.core.windows.net/key", "io/gocloud/azure"}, + } { + t.Run(tt.location, func(t *testing.T) { + _, err := icebergio.LoadFS(context.Background(), nil, tt.location) + require.ErrorIs(t, err, icebergio.ErrIOSchemeNotFound) + assert.ErrorContains(t, err, tt.wantHint) + }) + } +} diff --git a/internal/schemes/schemes.go b/internal/schemes/schemes.go index c3810b068..b7274132b 100644 --- a/internal/schemes/schemes.go +++ b/internal/schemes/schemes.go @@ -19,24 +19,26 @@ package schemes import "slices" +// S3 covers oss because Alibaba OSS is reached through its S3-compatible API and the AWS SDK. +// Unlike Java, which has a dedicated OSSFileIO. +// OSS users here configure s3.* credential keys, not oss.*. var ( S3 = []string{"s3", "s3a", "s3n", "oss"} GCS = []string{"gs"} Azure = []string{"abfs", "abfss", "wasb", "wasbs"} ) -var byBackend = map[string][]string{ - "s3": S3, - "gcs": GCS, - "azure": Azure, -} - +// BackendFor returns the io/gocloud subpackage that registers scheme, or +// an empty string if no backend claims it. func BackendFor(scheme string) string { - for backend, list := range byBackend { - if slices.Contains(list, scheme) { - return backend - } + switch { + case slices.Contains(S3, scheme): + return "s3" + case slices.Contains(GCS, scheme): + return "gcs" + case slices.Contains(Azure, scheme): + return "azure" + default: + return "" } - - return "" } diff --git a/io/gocloud/azure/azure.go b/io/gocloud/azure/azure.go index 71bcad919..06650a6fd 100644 --- a/io/gocloud/azure/azure.go +++ b/io/gocloud/azure/azure.go @@ -109,8 +109,8 @@ func newAdlsLocation(adlsURI *url.URL) (*adlsLocation, error) { // Construct a Azure bucket from a URL func createAzureBucket(ctx context.Context, parsed *url.URL, props map[string]string) (*blob.Bucket, error) { - adlsSasTokens := blobfs.PropertiesWithPrefix(props, io.ADLSSasTokenPrefix) - adlsConnectionStrings := blobfs.PropertiesWithPrefix(props, io.ADLSConnectionStringPrefix) + adlsSasTokens := propertiesWithPrefix(props, io.ADLSSasTokenPrefix) + adlsConnectionStrings := propertiesWithPrefix(props, io.ADLSConnectionStringPrefix) // Construct the client location, err := newAdlsLocation(parsed) diff --git a/io/gocloud/azure/register_test.go b/io/gocloud/azure/register_test.go index 96fb5f4ce..97e0102ec 100644 --- a/io/gocloud/azure/register_test.go +++ b/io/gocloud/azure/register_test.go @@ -29,7 +29,13 @@ import ( ) func TestRegistersOnlyItsOwnSchemes(t *testing.T) { - assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.Azure...), io.GetRegisteredSchemes()) + registered := io.GetRegisteredSchemes() + assert.Subset(t, registered, schemes.Azure) + for _, list := range [][]string{schemes.S3, schemes.GCS} { + for _, scheme := range list { + assert.NotContains(t, registered, scheme) + } + } } func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { @@ -37,7 +43,7 @@ func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { for _, tt := range []struct { location string - errValue string + wantHint string }{ {"s3://bucket/key", "io/gocloud/s3"}, {"s3a://bucket/key", "io/gocloud/s3"}, @@ -46,7 +52,7 @@ func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { t.Run(tt.location, func(t *testing.T) { _, err := io.LoadFS(ctx, nil, tt.location) require.ErrorIs(t, err, io.ErrIOSchemeNotFound) - assert.ErrorContains(t, err, tt.errValue) + assert.ErrorContains(t, err, tt.wantHint) }) } } diff --git a/io/gocloud/blobfs/utils.go b/io/gocloud/azure/utils.go similarity index 92% rename from io/gocloud/blobfs/utils.go rename to io/gocloud/azure/utils.go index 9c5315bc8..26a314e2b 100644 --- a/io/gocloud/blobfs/utils.go +++ b/io/gocloud/azure/utils.go @@ -15,11 +15,11 @@ // specific language governing permissions and limitations // under the License. -package blobfs +package azure import "strings" -func PropertiesWithPrefix(props map[string]string, prefix string) map[string]string { +func propertiesWithPrefix(props map[string]string, prefix string) map[string]string { result := map[string]string{} for k, v := range props { if after, ok := strings.CutPrefix(k, prefix); ok { diff --git a/io/gocloud/blobfs/blob.go b/io/gocloud/blobfs/blob.go index 0a1010da0..9406ffaf9 100644 --- a/io/gocloud/blobfs/blob.go +++ b/io/gocloud/blobfs/blob.go @@ -86,6 +86,8 @@ var ErrEmptyObjectKey = errors.New("object key is empty") // URI to access it; this backend does not route across authorities. var ErrUnsupportedObjectAuthority = errors.New("object URI authority is not supported by this FileIO") +// ObjectLocation is a parsed object-store URI: +// the scheme, the authority owning the object, and the object key within that authority. type ObjectLocation struct { scheme string authority string @@ -125,6 +127,19 @@ func splitObjectLocation(location string) (ObjectLocation, error) { return NewObjectLocation(scheme, authority, key), nil } +// Scheme returns the URI scheme, empty for a location parsed from a bare key. +func (o ObjectLocation) Scheme() string { return o.scheme } + +// Authority returns the bucket or container owning the object, +// empty for a location parsed from a bare key. +func (o ObjectLocation) Authority() string { return o.authority } + +// Key returns the object key within the authority. +func (o ObjectLocation) Key() string { return o.key } + +// NewObjectLocation builds a location for a URI carrying an explicit authority. +// Backends parsing their own URI grammar use it so the URI prefix stays +// consistent with the scheme and authority they parsed. func NewObjectLocation(scheme, authority, key string) ObjectLocation { return ObjectLocation{ scheme: scheme, @@ -135,8 +150,12 @@ func NewObjectLocation(scheme, authority, key string) ObjectLocation { } } +// ObjectLocationExtractor resolves an input path to the object location this +// FileIO acts on, rejecting paths belonging to another authority. type ObjectLocationExtractor func(location string) (ObjectLocation, error) +// KeyExtractorFromObjectLocation adapts an extractor to a KeyExtractor by +// keeping only the object key. func KeyExtractorFromObjectLocation(extract ObjectLocationExtractor) KeyExtractor { return func(location string) (string, error) { parsed, err := extract(location) @@ -148,6 +167,8 @@ func KeyExtractorFromObjectLocation(extract ObjectLocationExtractor) KeyExtracto } } +// DefaultObjectLocationExtractor extracts locations from URIs of the form scheme://bucket/key, +// accepting only the given schemes and bucket. func DefaultObjectLocationExtractor(bucketName string, allowedSchemes ...string) ObjectLocationExtractor { return func(location string) (ObjectLocation, error) { parsed, err := splitObjectLocation(location) @@ -183,6 +204,7 @@ func defaultKeyExtractor(bucketName string, allowedSchemes ...string) KeyExtract return KeyExtractorFromObjectLocation(DefaultObjectLocationExtractor(bucketName, allowedSchemes...)) } +// FileIO is the iceberg-go/io implementation backed by a gocloud.dev bucket. type FileIO struct { *blob.Bucket @@ -285,13 +307,12 @@ func (bfs *FileIO) Open(path string) (icebergio.File, error) { } func (bfs *FileIO) Remove(name string) error { - var err error - name, err = bfs.preprocess(name) + key, err := bfs.preprocess(name) if err != nil { return &fs.PathError{Op: "remove", Path: name, Err: err} } - if err := bfs.Delete(bfs.ctx, name); err != nil { + if err := bfs.Delete(bfs.ctx, key); err != nil { if gcerrors.Code(err) == gcerrors.NotFound { marker := directoryMarker(name) if marker != "" { @@ -316,13 +337,12 @@ func (bfs *FileIO) Create(name string) (icebergio.FileWriter, error) { } func (bfs *FileIO) WriteFile(name string, content []byte) error { - var err error - name, err = bfs.preprocess(name) + key, err := bfs.preprocess(name) if err != nil { return &fs.PathError{Op: "write file", Path: name, Err: err} } - return bfs.WriteAll(bfs.ctx, name, content, nil) + return bfs.WriteAll(bfs.ctx, key, content, nil) } // NewWriter returns a Writer that writes to the blob stored at path. @@ -333,17 +353,17 @@ func (bfs *FileIO) WriteFile(name string, content []byte) error { // // The caller must call Close on the returned Writer, even if the write is // aborted. -func (bfs *FileIO) NewWriter(ctx context.Context, path string, overwrite bool, opts *blob.WriterOptions) (w *blobWriteFile, err error) { - path, err = bfs.preprocess(path) +func (bfs *FileIO) NewWriter(ctx context.Context, path string, overwrite bool, opts *blob.WriterOptions) (w *Writer, err error) { + key, err := bfs.preprocess(path) if err != nil { return nil, &fs.PathError{Op: "new writer", Path: path, Err: err} } - if !fs.ValidPath(path) { + if !fs.ValidPath(key) { return nil, &fs.PathError{Op: "new writer", Path: path, Err: fs.ErrInvalid} } if !overwrite { - if exists, err := bfs.Exists(ctx, path); err != nil || exists { + if exists, err := bfs.Exists(ctx, key); err != nil || exists { if err != nil { return nil, &fs.PathError{Op: "new writer", Path: path, Err: err} } @@ -351,18 +371,20 @@ func (bfs *FileIO) NewWriter(ctx context.Context, path string, overwrite bool, o return nil, &fs.PathError{Op: "new writer", Path: path, Err: fs.ErrInvalid} } } - bw, err := bfs.Bucket.NewWriter(ctx, path, opts) + bw, err := bfs.Bucket.NewWriter(ctx, key, opts) if err != nil { return nil, err } - return &blobWriteFile{ + return &Writer{ Writer: bw, - name: path, + name: key, }, nil } +// New returns a FileIO backed by bucket, +// using extractObject to map input paths to object keys within it. func New(ctx context.Context, bucket *blob.Bucket, extractObject ObjectLocationExtractor) *FileIO { return &FileIO{Bucket: bucket, extractObject: extractObject, ctx: ctx} } @@ -688,13 +710,15 @@ func (bfs *FileIO) RemoveAll(name string) error { return nil } -type blobWriteFile struct { +// Writer is the FileWriter returned by NewWriter. +// It exposes the underlying gocloud.dev writer for callers that need its options. +type Writer struct { *blob.Writer name string b *FileIO } -func (f *blobWriteFile) Name() string { return f.name } -func (f *blobWriteFile) Sys() any { return f.b } -func (f *blobWriteFile) Close() error { return f.Writer.Close() } -func (f *blobWriteFile) Write(p []byte) (int, error) { return f.Writer.Write(p) } +func (f *Writer) Name() string { return f.name } +func (f *Writer) Sys() any { return f.b } +func (f *Writer) Close() error { return f.Writer.Close() } +func (f *Writer) Write(p []byte) (int, error) { return f.Writer.Write(p) } diff --git a/io/gocloud/blobfs/blob_test.go b/io/gocloud/blobfs/blob_test.go index 609b8d08c..32eb9972c 100644 --- a/io/gocloud/blobfs/blob_test.go +++ b/io/gocloud/blobfs/blob_test.go @@ -883,3 +883,32 @@ func TestBlobFileIORemoveAll(t *testing.T) { require.NoError(t, bfs.RemoveAll("s3://test-bucket/missing")) } + +func TestBlobFileIOPreprocessErrorRetainsOriginalPath(t *testing.T) { + t.Parallel() + + bucket := memblob.OpenBucket(nil) + t.Cleanup(func() { require.NoError(t, bucket.Close()) }) + fileIO := testBlobFileIO(context.Background(), "my-bucket", bucket) + name := "s3://other-bucket/file.parquet" + + for _, tt := range []struct { + op string + run func() error + }{ + {"remove", func() error { return fileIO.Remove(name) }}, + {"write file", func() error { return fileIO.WriteFile(name, nil) }}, + {"new writer", func() error { + _, err := fileIO.NewWriter(context.Background(), name, true, nil) + + return err + }}, + } { + t.Run(tt.op, func(t *testing.T) { + var pathErr *fs.PathError + require.ErrorAs(t, tt.run(), &pathErr) + assert.Equal(t, tt.op, pathErr.Op) + assert.Equal(t, name, pathErr.Path) + }) + } +} diff --git a/io/gocloud/compat_test.go b/io/gocloud/compat_test.go index 4bd3a8219..f2134c4eb 100644 --- a/io/gocloud/compat_test.go +++ b/io/gocloud/compat_test.go @@ -21,6 +21,7 @@ import ( "context" "testing" + "github.com/apache/iceberg-go/internal/schemes" "github.com/apache/iceberg-go/io" "github.com/apache/iceberg-go/io/gocloud" "github.com/apache/iceberg-go/io/gocloud/blobfs" @@ -38,6 +39,13 @@ var ( _ error = gocloud.ErrUnsupportedObjectAuthority ) +func TestRegistersAllCloudSchemes(t *testing.T) { + registered := io.GetRegisteredSchemes() + for _, list := range [][]string{schemes.S3, schemes.GCS, schemes.Azure} { + assert.Subset(t, registered, list) + } +} + func TestDeprecatedParseConfigWrappers(t *testing.T) { ctx := context.Background() diff --git a/io/gocloud/gcs/register_test.go b/io/gocloud/gcs/register_test.go index 7897f082e..1c00ca1c5 100644 --- a/io/gocloud/gcs/register_test.go +++ b/io/gocloud/gcs/register_test.go @@ -29,7 +29,13 @@ import ( ) func TestRegistersOnlyItsOwnSchemes(t *testing.T) { - assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.GCS...), io.GetRegisteredSchemes()) + registered := io.GetRegisteredSchemes() + assert.Subset(t, registered, schemes.GCS) + for _, list := range [][]string{schemes.S3, schemes.Azure} { + for _, scheme := range list { + assert.NotContains(t, registered, scheme) + } + } } func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { diff --git a/io/gocloud/s3/register_test.go b/io/gocloud/s3/register_test.go index 54c6ca384..c380262e8 100644 --- a/io/gocloud/s3/register_test.go +++ b/io/gocloud/s3/register_test.go @@ -29,7 +29,13 @@ import ( ) func TestRegistersOnlyItsOwnSchemes(t *testing.T) { - assert.ElementsMatch(t, append([]string{"file", "", "mem"}, schemes.S3...), io.GetRegisteredSchemes()) + registered := io.GetRegisteredSchemes() + assert.Subset(t, registered, schemes.S3) + for _, list := range [][]string{schemes.GCS, schemes.Azure} { + for _, scheme := range list { + assert.NotContains(t, registered, scheme) + } + } } func TestOtherCloudSchemesRemainUnregistered(t *testing.T) { From 5c74dde4473a2f5e13142346c9720ebc7e8ae226 Mon Sep 17 00:00:00 2001 From: badalprasadsingh Date: Mon, 31 Aug 2026 09:13:30 +0530 Subject: [PATCH 3/3] fix: remove of fileio Signed-off-by: badalprasadsingh --- io/gocloud/blobfs/blob.go | 2 +- io/gocloud/blobfs/blob_test.go | 16 ++++++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/io/gocloud/blobfs/blob.go b/io/gocloud/blobfs/blob.go index 9406ffaf9..260c02912 100644 --- a/io/gocloud/blobfs/blob.go +++ b/io/gocloud/blobfs/blob.go @@ -314,7 +314,7 @@ func (bfs *FileIO) Remove(name string) error { if err := bfs.Delete(bfs.ctx, key); err != nil { if gcerrors.Code(err) == gcerrors.NotFound { - marker := directoryMarker(name) + marker := directoryMarker(key) if marker != "" { if markerErr := bfs.Delete(bfs.ctx, marker); markerErr == nil { return nil diff --git a/io/gocloud/blobfs/blob_test.go b/io/gocloud/blobfs/blob_test.go index 32eb9972c..cad387779 100644 --- a/io/gocloud/blobfs/blob_test.go +++ b/io/gocloud/blobfs/blob_test.go @@ -912,3 +912,19 @@ func TestBlobFileIOPreprocessErrorRetainsOriginalPath(t *testing.T) { }) } } + +func TestBlobFileIORemoveFallsBackToDirectoryMarker(t *testing.T) { + t.Parallel() + + ctx := context.Background() + bucket := memblob.OpenBucket(nil) + t.Cleanup(func() { require.NoError(t, bucket.Close()) }) + require.NoError(t, bucket.WriteAll(ctx, "ns/tbl/", nil, nil)) + + fileIO := testBlobFileIO(ctx, "test-bucket", bucket) + require.NoError(t, fileIO.Remove("s3://test-bucket/ns/tbl")) + + exists, err := bucket.Exists(ctx, "ns/tbl/") + require.NoError(t, err) + assert.False(t, exists) +}