This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new d6005444c refactor(io): support isolated scheme registries (#2291)
d6005444c is described below
commit d6005444cac93512f727ae24dafd7eeba88faf6e
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 27 15:00:09 2026 +0100
refactor(io): support isolated scheme registries (#2291)
---
io/gocloud/azure/register.go | 2 +-
io/gocloud/gcs/register.go | 2 +-
io/gocloud/s3/register.go | 2 +-
io/io.go | 33 +++----
io/mem.go | 2 +-
io/registry.go | 146 +++++++++++++++++++++++-------
io/registry_test.go | 205 +++++++++++++++++++++++++++++++++++++++++--
website/src/configuration.md | 18 +++-
8 files changed, 347 insertions(+), 63 deletions(-)
diff --git a/io/gocloud/azure/register.go b/io/gocloud/azure/register.go
index 1bb0c6c64..fa2b38b8d 100644
--- a/io/gocloud/azure/register.go
+++ b/io/gocloud/azure/register.go
@@ -42,6 +42,6 @@ func init() {
}
for _, scheme := range schemes.Azure {
- io.Register(scheme, factory)
+ io.MustRegister(scheme, factory)
}
}
diff --git a/io/gocloud/gcs/register.go b/io/gocloud/gcs/register.go
index f72d91d4f..9650c9bd2 100644
--- a/io/gocloud/gcs/register.go
+++ b/io/gocloud/gcs/register.go
@@ -42,6 +42,6 @@ func init() {
}
for _, scheme := range schemes.GCS {
- io.Register(scheme, factory)
+ io.MustRegister(scheme, factory)
}
}
diff --git a/io/gocloud/s3/register.go b/io/gocloud/s3/register.go
index 8e59b8dc4..6e2421ed4 100644
--- a/io/gocloud/s3/register.go
+++ b/io/gocloud/s3/register.go
@@ -42,6 +42,6 @@ func init() {
}
for _, scheme := range schemes.S3 {
- io.Register(scheme, factory)
+ io.MustRegister(scheme, factory)
}
}
diff --git a/io/io.go b/io/io.go
index b7e53a1db..71652b178 100644
--- a/io/io.go
+++ b/io/io.go
@@ -359,24 +359,21 @@ func (f ioFile) ReadDir(count int) ([]fs.DirEntry, error)
{
return d.ReadDir(count)
}
-// LoadFS takes a map of properties and an optional URI location
-// and attempts to infer an IO object from it using the registered
-// scheme factories.
-//
-// The scheme is extracted from the location URI and used to look up
-// the appropriate factory from the registry. The local filesystem
-// (file:// or empty scheme) is registered by default.
-//
-// Additional schemes can be registered by importing subpackages.
-// Import the backend for each cloud in use, or io/gocloud for all of them:
-//
-// import _ "github.com/apache/iceberg-go/io/gocloud/s3"
+// LoadFS takes properties and a URI location and resolves a FileIO using
+// the global scheme registry. When location is empty, the warehouse property
+// is used. Import the desired io/gocloud backends for their scheme factories.
func LoadFS(ctx context.Context, props map[string]string, location string)
(IO, error) {
+ return defaultRegistry.LoadFS(ctx, props, location)
+}
+
+// LoadFS loads a FileIO using only this registry's scheme factories.
+// If location is empty, the warehouse property is used instead.
+func (r *Registry) LoadFS(ctx context.Context, props map[string]string,
location string) (IO, error) {
if location == "" {
location = props["warehouse"]
}
- iofs, err := inferFileIOFromScheme(ctx, location, props)
+ iofs, err := r.inferFileIOFromScheme(ctx, location, props)
if err != nil {
return nil, err
}
@@ -388,10 +385,16 @@ func LoadFS(ctx context.Context, props map[string]string,
location string) (IO,
return iofs, nil
}
-// LoadFSFunc is a helper function to create IO Func factory for later usage
+// LoadFSFunc constructs a lazy factory backed by the global scheme registry.
func LoadFSFunc(props map[string]string, location string) func(ctx
context.Context) (IO, error) {
+ return defaultRegistry.LoadFSFunc(props, location)
+}
+
+// LoadFSFunc constructs a lazy factory backed by this registry. It keeps the
+// registry selection even when other schemes are registered globally later.
+func (r *Registry) LoadFSFunc(props map[string]string, location string)
func(ctx context.Context) (IO, error) {
return func(ctx context.Context) (IO, error) {
- iofs, err := LoadFS(ctx, props, location)
+ iofs, err := r.LoadFS(ctx, props, location)
if err != nil {
return nil, fmt.Errorf("failed to load metadata file at
%s: %w", location, err)
}
diff --git a/io/mem.go b/io/mem.go
index ef46346ba..8b00a7935 100644
--- a/io/mem.go
+++ b/io/mem.go
@@ -48,7 +48,7 @@ func openOrCreateMemBucket(bucket string) *MemFS {
}
func init() {
- Register("mem", func(_ context.Context, parsed *url.URL, _
map[string]string) (IO, error) {
+ MustRegister("mem", func(_ context.Context, parsed *url.URL, _
map[string]string) (IO, error) {
return openOrCreateMemBucket(parsed.Host), nil
})
}
diff --git a/io/registry.go b/io/registry.go
index 8bebcd2a2..1553bf620 100644
--- a/io/registry.go
+++ b/io/registry.go
@@ -19,6 +19,7 @@ package io
import (
"context"
+ "errors"
"fmt"
"maps"
"net/url"
@@ -28,53 +29,132 @@ import (
"github.com/apache/iceberg-go/internal/schemes"
)
-type registry map[string]SchemeFactory
+// SchemeFactory creates an IO implementation for a URI and its properties.
+type SchemeFactory func(ctx context.Context, parsed *url.URL, props
map[string]string) (IO, error)
var (
- regMutex sync.RWMutex
- defaultRegistry = registry{}
+ // ErrSchemeAlreadyRegistered indicates that a registry already has a
factory for a scheme.
+ ErrSchemeAlreadyRegistered = errors.New("io scheme already registered")
+ // ErrInvalidSchemeFactory indicates that an attempt was made to
register a nil factory.
+ ErrInvalidSchemeFactory = errors.New("io scheme factory is nil")
)
-// SchemeFactory is a function that creates an IO implementation for a given
URI and properties.
-type SchemeFactory func(ctx context.Context, parsed *url.URL, props
map[string]string) (IO, error)
+// Registry manages FileIO factories for URI schemes. A registry may be used by
+// multiple goroutines. Factories run without the registry lock held, so they
+// may register other schemes or load additional FileIO instances.
+type Registry struct {
+ mu sync.RWMutex
+ factories map[string]SchemeFactory
+}
-// Register adds a new scheme factory to the registry. If the scheme is
already registered, it will panic.
-func Register(scheme string, factory SchemeFactory) {
+func builtInRegistry() *Registry {
+ localFSFactory := func(context.Context, *url.URL, map[string]string)
(IO, error) {
+ return LocalFS{}, nil
+ }
+
+ return &Registry{factories: map[string]SchemeFactory{
+ "": localFSFactory,
+ "file": localFSFactory,
+ }}
+}
+
+// defaultRegistry is populated by side-effect imports of the cloud providers.
+// Independent registries clone these registrations rather than sharing the
map.
+var defaultRegistry = builtInRegistry()
+
+// NewRegistry creates an independent snapshot of the globally registered
+// schemes, including cloud providers already registered through side-effect
+// imports. Future registrations in either registry do not affect the other.
+// Called from a package-level var initializer, it can miss backends that the
+// calling package does not import.
+func NewRegistry() *Registry {
+ return defaultRegistry.Clone()
+}
+
+// Clone returns an independent copy of a registry's factory registrations.
+// Factories themselves are shared, but the registration maps and locks are
not.
+func (r *Registry) Clone() *Registry {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ return &Registry{factories: maps.Clone(r.factories)}
+}
+
+// Register adds a scheme factory. Unlike MustRegister, it returns errors for
+// invalid or duplicate registrations rather than panicking.
+func (r *Registry) Register(scheme string, factory SchemeFactory) error {
if factory == nil {
- panic("io: Register factory is nil")
+ return ErrInvalidSchemeFactory
}
- regMutex.Lock()
- defer regMutex.Unlock()
+ r.mu.Lock()
+ defer r.mu.Unlock()
- if _, dup := defaultRegistry[scheme]; dup {
- panic("io: Register called twice for scheme " + scheme)
+ if _, exists := r.factories[scheme]; exists {
+ return fmt.Errorf("%w: %q", ErrSchemeAlreadyRegistered, scheme)
}
- defaultRegistry[scheme] = factory
+
+ if r.factories == nil {
+ r.factories = make(map[string]SchemeFactory)
+ }
+ r.factories[scheme] = factory
+
+ return nil
}
-// Unregister removes the requested scheme factory from the registry.
-func Unregister(scheme string) {
- regMutex.Lock()
- defer regMutex.Unlock()
- delete(defaultRegistry, scheme)
+// MustRegister registers a scheme or panics. It is intended for provider init
+// functions where failing to register a required backend is unrecoverable.
+func (r *Registry) MustRegister(scheme string, factory SchemeFactory) {
+ if err := r.Register(scheme, factory); err != nil {
+ panic(err)
+ }
}
-// GetRegisteredSchemes returns registered scheme names in sorted order.
-func GetRegisteredSchemes() []string {
- regMutex.RLock()
- defer regMutex.RUnlock()
+// Unregister removes a scheme from this registry only.
+func (r *Registry) Unregister(scheme string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
- return slices.Sorted(maps.Keys(defaultRegistry))
+ delete(r.factories, scheme)
}
-func init() {
- // Register local filesystem schemes
- localFSFactory := func(ctx context.Context, parsed *url.URL, props
map[string]string) (IO, error) {
- return LocalFS{}, nil
+// Names returns the registered scheme names in sorted order.
+func (r *Registry) Names() []string {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ return slices.Sorted(maps.Keys(r.factories))
+}
+
+// Register adds a scheme factory to the global registry. As in earlier
versions,
+// a nil factory or duplicate scheme causes a panic.
+func Register(scheme string, factory SchemeFactory) {
+ if factory == nil {
+ panic("io: Register factory is nil")
}
- Register("file", localFSFactory)
- Register("", localFSFactory)
+ if err := defaultRegistry.Register(scheme, factory); err != nil {
+ if errors.Is(err, ErrSchemeAlreadyRegistered) {
+ panic("io: Register called twice for scheme " + scheme)
+ }
+ panic(err)
+ }
+}
+
+// MustRegister registers a global scheme or panics. Use this from backend init
+// functions. For registration errors instead of panics, create an independent
+// registry and call [Registry.Register].
+func MustRegister(scheme string, factory SchemeFactory) {
+ defaultRegistry.MustRegister(scheme, factory)
+}
+
+// Unregister removes a scheme from the global registry.
+func Unregister(scheme string) {
+ defaultRegistry.Unregister(scheme)
+}
+
+// GetRegisteredSchemes returns the names registered globally in sorted order.
+func GetRegisteredSchemes() []string {
+ return defaultRegistry.Names()
}
func schemeImportHint(backend string) string {
@@ -91,15 +171,15 @@ func schemeRegistrationHint(scheme string) string {
return schemeImportHint(backend)
}
-func inferFileIOFromScheme(ctx context.Context, path string, props
map[string]string) (IO, error) {
+func (r *Registry) inferFileIOFromScheme(ctx context.Context, path string,
props map[string]string) (IO, error) {
parsed, err := url.Parse(path)
if err != nil {
return nil, err
}
- regMutex.RLock()
- factory, ok := defaultRegistry[parsed.Scheme]
- regMutex.RUnlock()
+ r.mu.RLock()
+ factory, ok := r.factories[parsed.Scheme]
+ r.mu.RUnlock()
if !ok {
hint := schemeRegistrationHint(parsed.Scheme)
diff --git a/io/registry_test.go b/io/registry_test.go
index 47a27c200..d10260477 100644
--- a/io/registry_test.go
+++ b/io/registry_test.go
@@ -19,7 +19,11 @@ package io_test
import (
"context"
+ "errors"
+ "fmt"
"net/url"
+ "slices"
+ "sync"
"testing"
"github.com/apache/iceberg-go/io"
@@ -97,14 +101,13 @@ func TestIORegistry(t *testing.T) {
assert.ErrorIs(t, err, io.ErrIOSchemeNotFound)
}
-func TestRegistryPanic(t *testing.T) {
- assert.PanicsWithValue(t, "io: Register factory is nil", func() {
- io.Register("invalid", nil)
- })
+func TestRegisterRejectsNilFactory(t *testing.T) {
+ assert.PanicsWithValue(t, "io: Register factory is nil", func() {
io.Register("invalid", nil) })
+ assert.NotContains(t, io.GetRegisteredSchemes(), "invalid")
}
-func TestRegisterDuplicatePanic(t *testing.T) {
- dummyFactory := func(ctx context.Context, parsed *url.URL, props
map[string]string) (io.IO, error) {
+func TestRegisterRejectsDuplicate(t *testing.T) {
+ dummyFactory := func(context.Context, *url.URL, map[string]string)
(io.IO, error) {
return io.LocalFS{}, nil
}
@@ -112,8 +115,41 @@ func TestRegisterDuplicatePanic(t *testing.T) {
defer io.Unregister("test-duplicate")
assert.PanicsWithValue(t, "io: Register called twice for scheme
test-duplicate", func() {
- io.Register("test-duplicate", dummyFactory)
- }, "Attempting to register the same scheme twice should panic")
+ io.Register("test-duplicate", func(context.Context, *url.URL,
map[string]string) (io.IO, error) {
+ return nil, assert.AnError
+ })
+ })
+ fs, err := io.LoadFS(t.Context(), nil, "test-duplicate://bucket/path")
+ require.NoError(t, err)
+ assert.IsType(t, io.LocalFS{}, fs)
+}
+
+func TestMustRegisterPanicsOnInvalidRegistration(t *testing.T) {
+ assert.Panics(t, func() { io.MustRegister("nil-factory-test", nil) })
+ assert.NotContains(t, io.GetRegisteredSchemes(), "nil-factory-test")
+}
+
+func TestScopedRegistryDuplicateDoesNotReplaceFactory(t *testing.T) {
+ t.Parallel()
+
+ reg := io.NewRegistry()
+ const scheme = "scoped-duplicate-regression"
+ require.NoError(t, reg.Register(scheme, func(context.Context, *url.URL,
map[string]string) (io.IO, error) {
+ return io.LocalFS{}, nil
+ }))
+
+ err := reg.Register(scheme, func(context.Context, *url.URL,
map[string]string) (io.IO, error) {
+ return nil, assert.AnError
+ })
+ require.ErrorIs(t, err, io.ErrSchemeAlreadyRegistered)
+
+ err = reg.Register("invalid-nil-regression", nil)
+ require.ErrorIs(t, err, io.ErrInvalidSchemeFactory)
+ assert.NotContains(t, reg.Names(), "invalid-nil-regression")
+
+ fs, err := reg.LoadFS(t.Context(), nil, scheme+"://bucket/key")
+ require.NoError(t, err)
+ assert.IsType(t, io.LocalFS{}, fs)
}
func TestLoadFS(t *testing.T) {
@@ -176,3 +212,156 @@ func TestLoadFSWithWarehouse(t *testing.T) {
assert.NotNil(t, iofs)
assert.IsType(t, io.LocalFS{}, iofs)
}
+
+func TestIndependentRegistryIsolation(t *testing.T) {
+ t.Parallel()
+
+ first := io.NewRegistry()
+ second := io.NewRegistry()
+ require.Equal(t, io.GetRegisteredSchemes(), first.Names())
+ require.Equal(t, first.Names(), second.Names())
+ assert.True(t, slices.IsSorted(first.Names()))
+
+ const scheme = "scoped-registry-test"
+ factory := func(context.Context, *url.URL, map[string]string) (io.IO,
error) {
+ return io.LocalFS{}, nil
+ }
+ require.NoError(t, first.Register(scheme, factory))
+
+ fs, err := first.LoadFS(t.Context(), nil, scheme+"://bucket/path")
+ require.NoError(t, err)
+ assert.IsType(t, io.LocalFS{}, fs)
+
+ _, err = second.LoadFS(t.Context(), nil, scheme+"://bucket/path")
+ require.ErrorIs(t, err, io.ErrIOSchemeNotFound)
+ _, err = io.LoadFS(t.Context(), nil, scheme+"://bucket/path")
+ require.ErrorIs(t, err, io.ErrIOSchemeNotFound)
+
+ clone := first.Clone()
+ first.Unregister(scheme)
+ _, err = first.LoadFS(t.Context(), nil, scheme+"://bucket/path")
+ require.ErrorIs(t, err, io.ErrIOSchemeNotFound)
+ _, err = clone.LoadFS(t.Context(), nil, scheme+"://bucket/path")
+ require.NoError(t, err)
+
+ _, err = first.LoadFS(t.Context(), nil, "file:///tmp/sample")
+ require.NoError(t, err)
+}
+
+func TestNewRegistrySnapshotsGlobalRegistrations(t *testing.T) {
+ const scheme = "late-global-registration-test"
+ before := io.NewRegistry()
+
+ io.Register(scheme, func(context.Context, *url.URL, map[string]string)
(io.IO, error) {
+ return io.LocalFS{}, nil
+ })
+ t.Cleanup(func() { io.Unregister(scheme) })
+
+ _, err := before.LoadFS(t.Context(), nil, scheme+"://bucket/key")
+ require.ErrorIs(t, err, io.ErrIOSchemeNotFound)
+
+ after := io.NewRegistry()
+ _, err = after.LoadFS(t.Context(), nil, scheme+"://bucket/key")
+ require.NoError(t, err)
+ assert.Contains(t, after.Names(), scheme)
+ assert.NotContains(t, before.Names(), scheme)
+}
+
+func TestRegistryZeroValueIsUsable(t *testing.T) {
+ var reg io.Registry
+ require.Empty(t, reg.Names())
+
+ factory := func(context.Context, *url.URL, map[string]string) (io.IO,
error) {
+ return io.LocalFS{}, nil
+ }
+ require.NoError(t, reg.Register("private", factory))
+
+ fs, err := reg.LoadFS(t.Context(), nil, "private://bucket/key")
+ require.NoError(t, err)
+ assert.IsType(t, io.LocalFS{}, fs)
+
+ cloned := reg.Clone()
+ reg.Unregister("private")
+ require.Empty(t, reg.Names())
+ assert.Equal(t, []string{"private"}, cloned.Names())
+}
+
+func TestRegistryAllowsFactoriesToRegisterSchemes(t *testing.T) {
+ t.Parallel()
+
+ reg := io.NewRegistry()
+ factory := func(context.Context, *url.URL, map[string]string) (io.IO,
error) {
+ err := reg.Register("second-scheme", func(context.Context,
*url.URL, map[string]string) (io.IO, error) {
+ return io.LocalFS{}, nil
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ return io.LocalFS{}, nil
+ }
+ require.NoError(t, reg.Register("first-scheme", factory))
+ _, err := reg.LoadFS(t.Context(), nil, "first-scheme://bucket/key")
+ require.NoError(t, err)
+ _, err = reg.LoadFS(t.Context(), nil, "second-scheme://bucket/key")
+ require.NoError(t, err)
+}
+
+func TestRegistryPreservesFactoryErrors(t *testing.T) {
+ t.Parallel()
+
+ reg := io.NewRegistry()
+ want := errors.New("factory failed")
+ require.NoError(t, reg.Register("fails", func(context.Context,
*url.URL, map[string]string) (io.IO, error) {
+ return nil, want
+ }))
+ _, err := reg.LoadFS(t.Context(), nil, "fails://bucket/key")
+ require.ErrorIs(t, err, want)
+}
+
+func TestRegistryLoadFSFuncUsesScopedFactories(t *testing.T) {
+ t.Parallel()
+
+ reg := io.NewRegistry()
+ require.NoError(t, reg.Register("lazy-scheme", func(context.Context,
*url.URL, map[string]string) (io.IO, error) {
+ return io.LocalFS{}, nil
+ }))
+ load := reg.LoadFSFunc(map[string]string{"warehouse":
"lazy-scheme://bucket/key"}, "")
+ _, err := load(t.Context())
+ require.NoError(t, err)
+ _, err = io.LoadFS(t.Context(), nil, "lazy-scheme://bucket/key")
+ require.ErrorIs(t, err, io.ErrIOSchemeNotFound)
+}
+
+func TestRegistryConcurrentRegistrationsAndLoads(t *testing.T) {
+ t.Parallel()
+
+ reg := io.NewRegistry()
+ const workers = 32
+ factory := func(context.Context, *url.URL, map[string]string) (io.IO,
error) {
+ return io.LocalFS{}, nil
+ }
+
+ var wg sync.WaitGroup
+ errs := make(chan error, workers)
+ for i := range workers {
+ wg.Go(func() {
+ scheme := fmt.Sprintf("worker-%d", i)
+ if err := reg.Register(scheme, factory); err != nil {
+ errs <- err
+
+ return
+ }
+
+ if _, err := reg.LoadFS(t.Context(), nil,
scheme+"://bucket/key"); err != nil {
+ errs <- err
+ }
+ })
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ require.NoError(t, err)
+ }
+ require.Len(t, reg.Names(), workers+len(io.NewRegistry().Names()))
+}
diff --git a/website/src/configuration.md b/website/src/configuration.md
index d565061b2..bda15dcb0 100644
--- a/website/src/configuration.md
+++ b/website/src/configuration.md
@@ -295,7 +295,7 @@ Two registries are user-extensible. The third
(LocationProvider) is currently in
### IO scheme registry
-Register a custom URL scheme with
[`io.Register`](https://github.com/apache/iceberg-go/blob/main/io/registry.go):
+Register a custom URL scheme from an `init` function with
[`io.MustRegister`](https://github.com/apache/iceberg-go/blob/main/io/registry.go):
```go
import (
@@ -306,17 +306,29 @@ import (
)
func init() {
- io.Register("myfs", func(ctx context.Context, parsed *url.URL, props
map[string]string) (io.IO, error) {
+ io.MustRegister("myfs", func(ctx context.Context, parsed *url.URL, props
map[string]string) (io.IO, error) {
return newMyFS(parsed, props)
})
}
```
-`io.Register` panics on `nil` factory or duplicate scheme. Built-in schemes:
`file`, `""` (the empty scheme).
+`io.Register` and `io.MustRegister` both panic for invalid or duplicate global
schemes, preserving the existing registration contract. `registry.Register` on
an independent registry returns `ErrInvalidSchemeFactory` or
`ErrSchemeAlreadyRegistered` instead. 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.
+To register schemes at runtime without changing other users of the same
process, create an independent registry (catalogs and `view.CreateView` still
resolve through the default registry):
+
+```go
+registry := io.NewRegistry()
+if err := registry.Register("myfs", myFactory); err != nil {
+ return err
+}
+fs, err := registry.LoadFS(ctx, props, "myfs://bucket/key")
+```
+
+`io.NewRegistry()` copies schemes already registered globally (including
blank-imported cloud backends) but keeps later registrations independent.
`registry.Clone()` makes an isolated copy of another registry.
`registry.Names()` lists the schemes registered in that instance.
`registry.LoadFSFunc()` is the lazy factory equivalent. Global `io.LoadFS()`
and `io.LoadFSFunc()` continue using the default registry.
+
### Catalog type registry
Register a custom catalog type with
[`catalog.Register`](https://github.com/apache/iceberg-go/blob/main/catalog/registry.go):
