Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions catalog/hadoop/hadoop.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down
4 changes: 2 additions & 2 deletions catalog/hadoop/io.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd hold on this import swap until we decide what happens to scheme registration for catalog/hadoop users.

Today the value import of io/gocloud pulls in that package's init(), so anyone importing catalog/hadoop gets all nine cloud schemes registered for free. Swapping to blobfs (which has no registration init()) means a program that imports only catalog/hadoop and talks to s3/gcs/azure will now build fine and then fail at the first LoadFS with ErrIOSchemeNotFound. That's a silent runtime regression on go get -u, with no compile-time signal.

I think the split itself is right, so I wouldn't want to re-link all three SDKs into hadoop just to preserve the old behavior. But we shouldn't let it break silently either. Either keep a blank import of io/gocloud here, or treat this as an intentional break and call it out loudly in the CHANGELOG and package doc (the website config page alone won't reach someone upgrading). wdyt?

)

// HadoopCatalogFS represents all the interfaces that a filesystem implementation
Expand All @@ -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)
58 changes: 58 additions & 0 deletions catalog/hadoop/io_test.go
Original file line number Diff line number Diff line change
@@ -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)
})
}
}
44 changes: 44 additions & 0 deletions internal/schemes/schemes.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
// 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"

// 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"}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth a doc comment on this slice: oss routes through the AWS S3 SDK here, which works via OSS's S3-compatible API but diverges from Java (dedicated OSSFileIO) and means OSS users configure s3.* credential keys, not oss.*. Easy to trip over otherwise.

GCS = []string{"gs"}
Azure = []string{"abfs", "abfss", "wasb", "wasbs"}
)

// BackendFor returns the io/gocloud subpackage that registers scheme, or
// an empty string if no backend claims it.
func BackendFor(scheme string) string {
switch {
case slices.Contains(S3, scheme):
return "s3"
case slices.Contains(GCS, scheme):
return "gcs"
case slices.Contains(Azure, scheme):
return "azure"
default:
return ""
}
}
25 changes: 10 additions & 15 deletions io/gocloud/azure.go → io/gocloud/azure/azure.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.

package gocloud
package azure

import (
"context"
Expand All @@ -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"
)
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

//go:build integration

package gocloud_test
package azure_test

import (
"context"
Expand All @@ -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"
Expand Down
80 changes: 75 additions & 5 deletions io/gocloud/azure_test.go → io/gocloud/azure/azure_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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,
},
{
Expand All @@ -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,
},
{
Expand All @@ -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 {
Expand All @@ -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)
}
47 changes: 47 additions & 0 deletions io/gocloud/azure/register.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
Loading
Loading