refactor(io): use the registry pattern for IO schemes

.github/workflows/go-integration.yml

@@ -65,9 +65,9 @@ jobs:
         env:
           AWS_S3_ENDPOINT: "${{ env.AWS_S3_ENDPOINT }}"
           AWS_REGION: "us-east-1"
-        run: |          
+        run: |
           go test -tags integration -v -run="^TestScanner" ./table
-          go test -tags integration -v ./io
+          go test -tags integration -v ./io/...
           go test -tags integration -v -run="^TestRestIntegration$" ./catalog/rest
           go test -tags=integration -v ./catalog/hive/...
       - name: Run spark integration tests

catalog/glue/glue_test.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
+	_ "github.com/apache/iceberg-go/io/gocloud"
 	"github.com/apache/iceberg-go/table"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"

catalog/rest/rest_integration_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/rest"
 	"github.com/apache/iceberg-go/io"
+	_ "github.com/apache/iceberg-go/io/gocloud"
 	"github.com/apache/iceberg-go/table"
 	"github.com/apache/iceberg-go/view"
 	"github.com/stretchr/testify/require"

catalog/sql/sql_integration_test.go

@@ -34,6 +34,7 @@ import (
 	"github.com/apache/iceberg-go/catalog"
 	"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/table"
 	"github.com/stretchr/testify/suite"
 )

catalog/sql/sql_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/internal"
 	sqlcat "github.com/apache/iceberg-go/catalog/sql"
+	_ "github.com/apache/iceberg-go/io/gocloud"
 	"github.com/apache/iceberg-go/table"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"

cmd/iceberg/main.go

@@ -31,6 +31,7 @@ import (
 	"github.com/apache/iceberg-go/catalog/hive"
 	"github.com/apache/iceberg-go/catalog/rest"
 	"github.com/apache/iceberg-go/config"
+	_ "github.com/apache/iceberg-go/io/gocloud"
 	"github.com/apache/iceberg-go/table"
 
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"

io/config.go

@@ -0,0 +1,54 @@
+// 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 io
+
+// Constants for S3 configuration options
+const (
+	S3Region                 = "s3.region"
+	S3SessionToken           = "s3.session-token"
+	S3SecretAccessKey        = "s3.secret-access-key"
+	S3AccessKeyID            = "s3.access-key-id"
+	S3EndpointURL            = "s3.endpoint"
+	S3ProxyURI               = "s3.proxy-uri"
+	S3ConnectTimeout         = "s3.connect-timeout"
+	S3SignerURI              = "s3.signer.uri"
+	S3ForceVirtualAddressing = "s3.force-virtual-addressing"
+)
+
+// Constants for GCS configuration options
+const (
+	GCSEndpoint   = "gcs.endpoint"
+	GCSKeyPath    = "gcs.keypath"
+	GCSJSONKey    = "gcs.jsonkey"
+	GCSCredType   = "gcs.credtype"
+	GCSUseJSONAPI = "gcs.usejsonapi" // set to anything to enable
+)
+
+// Constants for Azure configuration options
+const (
+	ADLSSasTokenPrefix         = "adls.sas-token."
+	ADLSConnectionStringPrefix = "adls.connection-string."
+	ADLSSharedKeyAccountName   = "adls.auth.shared-key.account.name"
+	ADLSSharedKeyAccountKey    = "adls.auth.shared-key.account.key"
+	ADLSEndpoint               = "adls.endpoint"
+	ADLSProtocol               = "adls.protocol"
+
+	// Not in use yet
+	// ADLSReadBlockSize          = "adls.read.block-size-bytes"
+	// ADLSWriteBlockSize         = "adls.write.block-size-bytes"
+)

io/azure.go → io/gocloud/azure.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package io
+package gocloud
 
 import (
 	"context"
@@ -28,6 +28,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"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"
 	"gocloud.dev/blob"
 	"gocloud.dev/blob/azureblob"
 )
@@ -36,20 +37,6 @@ import (
 // https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/azure/src/main/java/org/apache/iceberg/azure/adlsv2/ADLSLocation.java#L47
 var adlsURIPattern = regexp.MustCompile(`^(abfss?|wasbs?)://([^/?#]+)(.*)?$`)
 
-// Constants for Azure configuration options
-const (
-	AdlsSasTokenPrefix         = "adls.sas-token."
-	AdlsConnectionStringPrefix = "adls.connection-string."
-	AdlsSharedKeyAccountName   = "adls.auth.shared-key.account.name"
-	AdlsSharedKeyAccountKey    = "adls.auth.shared-key.account.key"
-	AdlsEndpoint               = "adls.endpoint"
-	AdlsProtocol               = "adls.protocol"
-
-	// Not in use yet
-	// AdlsReadBlockSize          = "adls.read.block-size-bytes"
-	// AdlsWriteBlockSize         = "adls.write.block-size-bytes"
-)
-
 // adlsLocation represents the parsed components of an Azure Data Lake Storage URI
 type adlsLocation struct {
 	accountName   string // Azure storage account name
@@ -119,25 +106,25 @@ 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, AdlsSasTokenPrefix)
-	adlsConnectionStrings := propertiesWithPrefix(props, AdlsConnectionStringPrefix)
+	adlsSasTokens := propertiesWithPrefix(props, io.ADLSSasTokenPrefix)
+	adlsConnectionStrings := propertiesWithPrefix(props, io.ADLSConnectionStringPrefix)
 
 	// Construct the client
 	location, err := newAdlsLocation(parsed)
 	if err != nil {
 		return nil, err
 	}
 
-	sharedKeyAccountName := props[AdlsSharedKeyAccountName]
-	endpoint := props[AdlsEndpoint]
-	protocol := props[AdlsProtocol]
+	sharedKeyAccountName := props[io.ADLSSharedKeyAccountName]
+	endpoint := props[io.ADLSEndpoint]
+	protocol := props[io.ADLSProtocol]
 
 	var client *container.Client
 
 	if sharedKeyAccountName != "" {
-		sharedKeyAccountKey, ok := props[AdlsSharedKeyAccountKey]
+		sharedKeyAccountKey, ok := props[io.ADLSSharedKeyAccountKey]
 		if !ok || sharedKeyAccountKey == "" {
-			return nil, fmt.Errorf("azure authentication: shared-key requires both %s and %s", AdlsSharedKeyAccountName, AdlsSharedKeyAccountKey)
+			return nil, fmt.Errorf("azure authentication: shared-key requires both %s and %s", io.ADLSSharedKeyAccountName, io.ADLSSharedKeyAccountKey)
 		}
 
 		containerURL, err := createContainerURL(location.accountName, protocol, endpoint, "", location.containerName)

io/azure_integration_test.go → io/gocloud/azure_integration_test.go

@@ -17,7 +17,7 @@
 
 //go:build integration
 
-package io_test
+package gocloud_test
 
 import (
 	"context"
@@ -30,6 +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/stretchr/testify/suite"
 	"github.com/uptrace/bun/driver/sqliteshim"
 	"gocloud.dev/blob/azureblob"
@@ -64,10 +65,10 @@ func (s *AzureBlobIOTestSuite) TestAzureBlobWarehouseKey() {
 		sqlcat.DriverKey:            sqliteshim.ShimName,
 		sqlcat.DialectKey:           string(sqlcat.SQLite),
 		"type":                      "sql",
-		io.AdlsSharedKeyAccountName: accountName,
-		io.AdlsSharedKeyAccountKey:  accountKey,
-		io.AdlsEndpoint:             endpoint,
-		io.AdlsProtocol:             protocol,
+		io.ADLSSharedKeyAccountName: accountName,
+		io.ADLSSharedKeyAccountKey:  accountKey,
+		io.ADLSEndpoint:             endpoint,
+		io.ADLSProtocol:             protocol,
 	}
 
 	cat, err := catalog.Load(context.Background(), "default", properties)
@@ -99,7 +100,7 @@ func (s *AzureBlobIOTestSuite) TestAzuriteWarehouseConnectionString() {
 		sqlcat.DriverKey:  sqliteshim.ShimName,
 		sqlcat.DialectKey: string(sqlcat.SQLite),
 		"type":            "sql",
-		io.AdlsConnectionStringPrefix + accountName: connectionString,
+		io.ADLSConnectionStringPrefix + accountName: connectionString,
 	}
 
 	cat, err := catalog.Load(context.Background(), "default", properties)

io/azure_test.go → io/gocloud/azure_test.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package io
+package gocloud
 
 import (
 	"context"

io/blob.go → io/gocloud/blob.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package io
+package gocloud
 
 import (
 	"context"
@@ -26,6 +26,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	icebergio "github.com/apache/iceberg-go/io"
 	"gocloud.dev/blob"
 )
 
@@ -103,7 +104,7 @@ func (bfs *blobFileIO) preprocess(path string) (string, error) {
 	return bfs.keyExtractor(path)
 }
 
-func (bfs *blobFileIO) Open(path string) (File, error) {
+func (bfs *blobFileIO) Open(path string) (icebergio.File, error) {
 	var err error
 	path, err = bfs.preprocess(path)
 	if err != nil {
@@ -133,7 +134,7 @@ func (bfs *blobFileIO) Remove(name string) error {
 	return bfs.Delete(bfs.ctx, name)
 }
 
-func (bfs *blobFileIO) Create(name string) (FileWriter, error) {
+func (bfs *blobFileIO) Create(name string) (icebergio.FileWriter, error) {
 	return bfs.NewWriter(bfs.ctx, name, true, nil)
 }
 
@@ -185,7 +186,7 @@ func (bfs *blobFileIO) NewWriter(ctx context.Context, path string, overwrite boo
 		nil
 }
 
-func createBlobFS(ctx context.Context, bucket *blob.Bucket, keyExtractor KeyExtractor) IO {
+func createBlobFS(ctx context.Context, bucket *blob.Bucket, keyExtractor KeyExtractor) icebergio.IO {
 	return &blobFileIO{Bucket: bucket, keyExtractor: keyExtractor, ctx: ctx}
 }
 

io/blob_test.go → io/gocloud/blob_test.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package io
+package gocloud
 
 import (
 	"context"

io/gcs.go → io/gocloud/gcs.go

@@ -15,29 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package io
+package gocloud
 
 import (
 	"context"
 	"net/url"
 
 	"cloud.google.com/go/storage"
 
+	"github.com/apache/iceberg-go/io"
 	"gocloud.dev/blob"
 	"gocloud.dev/blob/gcsblob"
 	"gocloud.dev/gcp"
 	"google.golang.org/api/option"
 )
 
-// Constants for GCS configuration options
-const (
-	GCSEndpoint   = "gcs.endpoint"
-	GCSKeyPath    = "gcs.keypath"
-	GCSJSONKey    = "gcs.jsonkey"
-	GCSCredType   = "gcs.credtype"
-	GCSUseJsonAPI = "gcs.usejsonapi" // set to anything to enable
-)
-
 var allowedGCSCredTypes = map[string]option.CredentialsType{
 	"service_account":              option.ServiceAccount,
 	"authorized_user":              option.AuthorizedUser,
@@ -48,22 +40,22 @@ var allowedGCSCredTypes = map[string]option.CredentialsType{
 // ParseGCSConfig parses GCS properties and returns a configuration.
 func ParseGCSConfig(props map[string]string) *gcsblob.Options {
 	var o []option.ClientOption
-	if url := props[GCSEndpoint]; url != "" {
+	if url := props[io.GCSEndpoint]; url != "" {
 		o = append(o, option.WithEndpoint(url))
 	}
 	var credType option.CredentialsType
-	if key := props[GCSCredType]; key != "" {
+	if key := props[io.GCSCredType]; key != "" {
 		if ct, ok := allowedGCSCredTypes[key]; ok {
 			credType = ct
 		}
 	}
-	if key := props[GCSJSONKey]; key != "" {
+	if key := props[io.GCSJSONKey]; key != "" {
 		o = append(o, option.WithAuthCredentialsJSON(credType, []byte(key)))
 	}
-	if path := props[GCSKeyPath]; path != "" {
+	if path := props[io.GCSKeyPath]; path != "" {
 		o = append(o, option.WithAuthCredentialsFile(credType, path))
 	}
-	if _, ok := props[GCSUseJsonAPI]; ok {
+	if _, ok := props[io.GCSUseJSONAPI]; ok {
 		o = append(o, storage.WithJSONReads())
 	}
 

io/gcs_integration_test.go → io/gocloud/gcs_integration_test.go

@@ -17,7 +17,7 @@
 
 //go:build integration
 
-package io_test
+package gocloud_test
 
 import (
 	"bytes"
@@ -32,6 +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/stretchr/testify/suite"
 	"github.com/uptrace/bun/driver/sqliteshim"
 )
@@ -114,7 +115,7 @@ func (s *GCSIOTestSuite) TestGCSWarehouse() {
 		"type":            "sql",
 		"warehouse":       fmt.Sprintf("gs://%s/iceberg/", gcsBucketName),
 		io.GCSEndpoint:    fmt.Sprintf("http://%s/", gcsEndpoint),
-		io.GCSUseJsonAPI:  "true",
+		io.GCSUseJSONAPI:  "true",
 	}
 
 	cat, err := catalog.Load(context.Background(), "default", properties)